From 3c36c46faf4655deb0e0dbcc58adb1f4c01f5ebc Mon Sep 17 00:00:00 2001 From: david Date: Wed, 19 Aug 2026 00:07:04 +0000 Subject: [PATCH 1/4] refactor(benchsdk): split SDK into @benchsdk/api, @benchsdk/worker, and @benchsdk/client umbrella - Add @benchsdk/api: typed REST client and shared platform types. - Add @benchsdk/worker: runWorker runtime, BenchmarkReporter, metrics, participants. - Keep @benchsdk/client as a backwards-compatible umbrella that re-exports from the focused packages and exposes client.runWorker(). - Update pnpm workspace and lockfile for the two new packages. - Adjust public-api.contract.test.ts to parse the new barrel .d.ts shape with workspace re-exports. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- packages/benchsdk-api/package.json | 63 ++ packages/benchsdk-api/src/client.ts | 450 ++++++++++ packages/benchsdk-api/src/index.ts | 74 ++ packages/benchsdk-api/src/types.ts | 745 ++++++++++++++++ packages/benchsdk-api/tsconfig.json | 9 + packages/benchsdk-api/tsup.config.ts | 16 + packages/benchsdk-api/vitest.config.ts | 8 + packages/benchsdk-worker/package.json | 65 ++ packages/benchsdk-worker/src/index.ts | 19 + packages/benchsdk-worker/src/metrics.ts | 93 ++ packages/benchsdk-worker/src/participants.ts | 54 ++ packages/benchsdk-worker/src/reporter.ts | 238 +++++ packages/benchsdk-worker/src/worker.ts | 396 +++++++++ packages/benchsdk-worker/tsconfig.json | 9 + packages/benchsdk-worker/tsup.config.ts | 13 + packages/benchsdk-worker/vitest.config.ts | 8 + packages/benchsdk/package.json | 5 +- .../src/__tests__/public-api.contract.test.ts | 54 +- packages/benchsdk/src/client.ts | 832 +----------------- packages/benchsdk/src/metrics.ts | 98 +-- packages/benchsdk/src/participants.ts | 56 +- packages/benchsdk/src/reporter.ts | 247 +----- pnpm-lock.yaml | 59 ++ pnpm-workspace.yaml | 2 + 24 files changed, 2389 insertions(+), 1224 deletions(-) create mode 100644 packages/benchsdk-api/package.json create mode 100644 packages/benchsdk-api/src/client.ts create mode 100644 packages/benchsdk-api/src/index.ts create mode 100644 packages/benchsdk-api/src/types.ts create mode 100644 packages/benchsdk-api/tsconfig.json create mode 100644 packages/benchsdk-api/tsup.config.ts create mode 100644 packages/benchsdk-api/vitest.config.ts create mode 100644 packages/benchsdk-worker/package.json create mode 100644 packages/benchsdk-worker/src/index.ts create mode 100644 packages/benchsdk-worker/src/metrics.ts create mode 100644 packages/benchsdk-worker/src/participants.ts create mode 100644 packages/benchsdk-worker/src/reporter.ts create mode 100644 packages/benchsdk-worker/src/worker.ts create mode 100644 packages/benchsdk-worker/tsconfig.json create mode 100644 packages/benchsdk-worker/tsup.config.ts create mode 100644 packages/benchsdk-worker/vitest.config.ts diff --git a/packages/benchsdk-api/package.json b/packages/benchsdk-api/package.json new file mode 100644 index 00000000..de24cdef --- /dev/null +++ b/packages/benchsdk-api/package.json @@ -0,0 +1,63 @@ +{ + "name": "@benchsdk/api", + "version": "0.1.0", + "private": false, + "type": "module", + "description": "Typed REST API client and shared types for the ComputeSDK benchmarks platform", + "author": "Garrison", + "license": "MIT", + "main": "./dist/index.cjs", + "module": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js", + "require": "./dist/index.cjs" + } + }, + "files": [ + "dist", + "README.md" + ], + "scripts": { + "build": "tsup", + "clean": "rimraf dist", + "dev": "tsup --watch", + "lint": "eslint", + "prepare": "tsup", + "pretest": "tsup", + "test": "vitest run", + "test:coverage": "vitest run --coverage", + "test:watch": "vitest watch", + "typecheck": "tsc --noEmit" + }, + "keywords": [ + "computesdk", + "benchmark", + "api", + "orchestrator" + ], + "repository": { + "type": "git", + "url": "https://github.com/computesdk/benchmarks.git", + "directory": "packages/benchsdk-api" + }, + "homepage": "https://www.computesdk.com", + "bugs": { + "url": "https://github.com/computesdk/benchmarks/issues" + }, + "engines": { + "node": ">=18.0.0" + }, + "dependencies": {}, + "devDependencies": { + "@types/node": "^20.0.0", + "@vitest/coverage-v8": "^1.0.0", + "eslint": "^8.37.0", + "rimraf": "^5.0.0", + "tsup": "^8.0.0", + "typescript": "^5.0.0", + "vitest": "^1.0.0" + } +} diff --git a/packages/benchsdk-api/src/client.ts b/packages/benchsdk-api/src/client.ts new file mode 100644 index 00000000..f82b5e96 --- /dev/null +++ b/packages/benchsdk-api/src/client.ts @@ -0,0 +1,450 @@ +import type { + BenchmarkAssignment, + BenchmarkArtifact, + BenchmarkClient, + BenchmarkClientConfig, + BenchmarkParticipant, + BenchmarkResource, + BenchmarkRun, + BenchmarkResultsOverview, + BenchmarkResultsOverviewInput, + BenchmarkRunImports, + BenchmarkRunResults, + BenchmarkRunSummaryInput, + BenchmarkRunTaskResults, + BenchmarkRunTaskResultsInput, + BenchmarkRunTimeline, + BenchmarkRunTimelineInput, + BenchmarkRunWorker, + BenchmarkWorkerAttempt, + CreateWorkerArtifactInput, + CreateWorkerArtifactResponse, + ClaimWorkerInput, + CreateRunInput, + JsonObject, + SendTaskResultsInput, + PlanWorkersInput, + TaskResultRecord, + TaskResultsResponse, + RunProgress, + UpdateBenchmarkInput, + UpdateParticipantInput, + UpdateRunInput, + UpdateWorkerInput, + UpsertBenchmarkInput, + UpsertParticipantInput, + UploadWorkerArtifactInput, + WorkerConcurrencySample, + WorkerHeartbeatInput, +} from './types'; + +const DEFAULT_BASE_URL = 'https://platform.computesdk.com/api/v1'; +const MAX_TASK_RESULT_RECORDS = 5000; +const MAX_TASK_RECORD_STEPS = 100; +const MAX_HEARTBEAT_CONCURRENCY_SAMPLES = 20; + +export class BenchmarkApiError extends Error { + constructor( + message: string, + public readonly status: number, + public readonly body: string, + ) { + super(message); + this.name = 'BenchmarkApiError'; + } +} + +function trimTrailingSlash(value: string): string { + return value.replace(/\/+$/, ''); +} + +function encodePath(value: string): string { + return encodeURIComponent(value); +} + +function queryString(input: Record): string { + const params = new URLSearchParams(); + for (const [key, value] of Object.entries(input)) { + if (value !== undefined) params.set(key, String(value)); + } + const value = params.toString(); + return value ? `?${value}` : ''; +} + +function getApiKey(input?: string): string | undefined { + return input ?? (typeof process !== 'undefined' ? process.env.COMPUTESDK_ADMIN_API_KEY ?? process.env.COMPUTESDK_API_KEY : undefined); +} + +function getErrorCode(error: unknown): string { + if (error instanceof Error && 'code' in error && typeof (error as { code: unknown }).code === 'string' && (error as { code: string }).code) { + return (error as { code: string }).code; + } + if (error instanceof Error && error.name) return error.name; + return 'ERROR'; +} + +function validateTaskResults(input: SendTaskResultsInput): void { + if (input.records.length > MAX_TASK_RESULT_RECORDS) { + throw new Error(`Benchmark task result batches are limited to ${MAX_TASK_RESULT_RECORDS} records.`); + } + + for (const record of input.records) { + if ((record.steps?.length ?? 0) > MAX_TASK_RECORD_STEPS) { + throw new Error(`Benchmark task result records are limited to ${MAX_TASK_RECORD_STEPS} steps.`); + } + } +} + +function validateHeartbeat(input: WorkerHeartbeatInput): void { + const concurrency = input.concurrency ?? []; + if (concurrency.length > MAX_HEARTBEAT_CONCURRENCY_SAMPLES) { + throw new Error(`Benchmark heartbeat concurrency is limited to ${MAX_HEARTBEAT_CONCURRENCY_SAMPLES} samples.`); + } + + const steps = new Set(); + for (const sample of concurrency) { + if (steps.has(sample.step)) { + throw new Error(`Benchmark heartbeat concurrency step values must be unique per heartbeat.`); + } + steps.add(sample.step); + } +} + +function normalizeArtifacts(data: { items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }): BenchmarkArtifact[] { + return data.items ?? data.artifacts ?? []; +} + +function bodySizeBytes(body: UploadWorkerArtifactInput['body']): number | undefined { + if (typeof body === 'string') return new TextEncoder().encode(body).byteLength; + if (body instanceof Blob) return body.size; + if (body instanceof ArrayBuffer) return body.byteLength; + if (ArrayBuffer.isView(body)) return body.byteLength; + if (body instanceof URLSearchParams) return new TextEncoder().encode(body.toString()).byteLength; + return undefined; +} + +export function createBenchmarkClient(config: BenchmarkClientConfig = {}): BenchmarkClient { + const baseUrl = trimTrailingSlash(config.baseUrl ?? DEFAULT_BASE_URL); + const apiKey = getApiKey(config.apiKey); + const fetchImpl = config.fetch ?? (typeof fetch !== 'undefined' ? fetch : undefined); + + if (!fetchImpl) { + throw new Error('fetch is not available'); + } + const doFetch = fetchImpl; + + async function request(method: string, path: string, body?: JsonObject): Promise { + const headers: Record = { 'Content-Type': 'application/json' }; + if (apiKey) headers.Authorization = `Bearer ${apiKey}`; + + const response = await doFetch(`${baseUrl}${path}`, { + method, + headers, + body: body === undefined ? undefined : JSON.stringify(body), + }); + const text = await response.text(); + + if (!response.ok) { + throw new BenchmarkApiError( + `Benchmark API request failed: ${response.status} ${response.statusText}`, + response.status, + text, + ); + } + + return (text ? JSON.parse(text) : {}) as T; + } + + async function sendTaskResults(input: SendTaskResultsInput): Promise { + if (input.records.length === 0) { + return {}; + } + validateTaskResults(input); + + return request( + 'POST', + `/benchmarks/${encodePath(input.benchmarkSlug)}/runs/${encodePath(input.runId)}/workers/${encodePath(input.workerId)}/events`, + { + type: 'task_results', + attemptId: input.attemptId, + sequenceNumber: input.sequenceNumber, + isFinal: input.isFinal, + records: input.records as unknown as JsonObject[], + }, + ); + } + + async function updateWorkerLifecycle( + action: 'heartbeat' | 'complete' | 'fail' | 'release', + benchmarkSlug: string, + runId: string, + workerId: string, + attemptId: string, + extra?: JsonObject, + ): Promise<{ worker: BenchmarkRunWorker; attempt: BenchmarkWorkerAttempt }> { + return request<{ worker: BenchmarkRunWorker; attempt: BenchmarkWorkerAttempt }>( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/${action}`, + { attemptId, ...(extra ?? {}) }, + ); + } + + const client: BenchmarkClient = { + async upsertBenchmark(slug, input) { + const data = await request<{ benchmark: BenchmarkResource }>('PUT', `/benchmarks/${encodePath(slug)}`, input as unknown as JsonObject); + return data.benchmark; + }, + + async getBenchmark(slug) { + const data = await request<{ benchmark: BenchmarkResource }>('GET', `/benchmarks/${encodePath(slug)}`); + return data.benchmark; + }, + + async updateBenchmark(slug, input: UpdateBenchmarkInput) { + const data = await request<{ benchmark: BenchmarkResource }>('PATCH', `/benchmarks/${encodePath(slug)}`, input as unknown as JsonObject); + return data.benchmark; + }, + + async listBenchmarks() { + const data = await request<{ items?: BenchmarkResource[]; benchmarks?: BenchmarkResource[] }>('GET', '/benchmarks'); + return data.items ?? data.benchmarks ?? []; + }, + + async createRun(benchmarkSlug, input) { + return request<{ run: BenchmarkRun; participants: BenchmarkParticipant[]; organizationSlug: string }>( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs`, + input as unknown as JsonObject, + ); + }, + + async listRuns(benchmarkSlug) { + const data = await request<{ items: BenchmarkRun[] }>('GET', `/benchmarks/${encodePath(benchmarkSlug)}/runs`); + return data.items; + }, + + async getRun(benchmarkSlug, runId) { + const data = await request<{ run: BenchmarkRun }>('GET', `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}`); + return data.run; + }, + + async updateRun(benchmarkSlug, runId, input: UpdateRunInput) { + const data = await request<{ run: BenchmarkRun }>( + 'PATCH', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}`, + input as unknown as JsonObject, + ); + return data.run; + }, + + async upsertParticipant(benchmarkSlug, runId, participantSlug, input = {}) { + const data = await request<{ participant: BenchmarkParticipant }>( + 'PUT', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, + input as JsonObject, + ); + return data.participant; + }, + + async listParticipants(benchmarkSlug, runId) { + const data = await request<{ items?: BenchmarkParticipant[]; participants?: BenchmarkParticipant[] }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants`, + ); + return data.items ?? data.participants ?? []; + }, + + async getParticipant(benchmarkSlug, runId, participantSlug) { + const data = await request<{ participant: BenchmarkParticipant }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, + ); + return data.participant; + }, + + async updateParticipant(benchmarkSlug, runId, participantSlug, input: UpdateParticipantInput) { + const data = await request<{ participant: BenchmarkParticipant }>( + 'PATCH', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, + input as unknown as JsonObject, + ); + return data.participant; + }, + + async getRunProgress(benchmarkSlug, runId) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/progress`, + ); + }, + + async listWorkers(benchmarkSlug, runId, participantSlug) { + const data = await request<{ items?: BenchmarkRunWorker[]; workers?: BenchmarkRunWorker[] }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers`, + ); + return data.items ?? data.workers ?? []; + }, + + async planWorkers(benchmarkSlug, runId, participantSlug, input: PlanWorkersInput = {}) { + const data = await request<{ items?: BenchmarkRunWorker[]; workers?: BenchmarkRunWorker[] }>( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers`, + input as JsonObject, + ); + return data.items ?? data.workers ?? []; + }, + + async getWorker(benchmarkSlug, runId, workerId) { + const data = await request<{ worker: BenchmarkRunWorker }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}`, + ); + return data.worker; + }, + + async updateWorker(benchmarkSlug, runId, workerId, input: UpdateWorkerInput) { + const data = await request<{ worker: BenchmarkRunWorker }>( + 'PATCH', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}`, + input as unknown as JsonObject, + ); + return data.worker; + }, + + async claimWorker(benchmarkSlug, runId, participantSlug, input: ClaimWorkerInput = {}) { + const data = await request<{ assignment: BenchmarkAssignment | null }>( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers/claim`, + input as JsonObject, + ); + return data.assignment; + }, + + sendTaskResults, + + async heartbeatWorker(benchmarkSlug, runId, workerId, input: WorkerHeartbeatInput) { + validateHeartbeat(input); + const { attemptId, ...extra } = input; + if (extra.currentStep == null) { + delete extra.currentStep; + } + return updateWorkerLifecycle('heartbeat', benchmarkSlug, runId, workerId, attemptId, extra as JsonObject); + }, + + releaseWorker(benchmarkSlug, runId, workerId, attemptId) { + return updateWorkerLifecycle('release', benchmarkSlug, runId, workerId, attemptId); + }, + + completeWorker(benchmarkSlug, runId, workerId, attemptId) { + return updateWorkerLifecycle('complete', benchmarkSlug, runId, workerId, attemptId); + }, + + failWorker(benchmarkSlug, runId, workerId, attemptId, error) { + return updateWorkerLifecycle('fail', benchmarkSlug, runId, workerId, attemptId, { + errorCode: getErrorCode(error), + errorMessage: error instanceof Error ? error.message : String(error ?? 'Unknown error'), + }); + }, + + async createWorkerArtifact(benchmarkSlug, runId, workerId, input: CreateWorkerArtifactInput) { + return request( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/artifacts`, + input as unknown as JsonObject, + ); + }, + + async uploadWorkerArtifact(benchmarkSlug, runId, workerId, input: UploadWorkerArtifactInput) { + const sizeBytes = bodySizeBytes(input.body); + const artifactInput: CreateWorkerArtifactInput = { + attemptId: input.attemptId, + kind: input.kind, + contentType: input.contentType, + name: input.name, + metadata: sizeBytes === undefined ? input.metadata : { ...input.metadata, sizeBytes }, + }; + const response = await client.createWorkerArtifact(benchmarkSlug, runId, workerId, artifactInput); + const uploadUrl = response.uploadUrl ?? response.artifact?.uploadUrl; + if (!uploadUrl) { + throw new Error('Benchmark artifact upload URL is missing.'); + } + const uploadResponse = await doFetch(uploadUrl, { + method: 'PUT', + headers: input.contentType ? { 'Content-Type': input.contentType } : undefined, + body: input.body, + }); + if (!uploadResponse.ok) { + const errorBody = await uploadResponse.text().catch(() => ''); + throw new BenchmarkApiError( + `Benchmark artifact upload failed with ${uploadResponse.status}`, + uploadResponse.status, + errorBody, + ); + } + return response; + }, + + async listRunArtifacts(benchmarkSlug, runId) { + const data = await request<{ items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/artifacts`, + ); + return normalizeArtifacts(data); + }, + + async listWorkerArtifacts(benchmarkSlug, runId, workerId) { + const data = await request<{ items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }>( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/artifacts`, + ); + return normalizeArtifacts(data); + }, + + async getBenchmarkResults(benchmarkSlug, input: BenchmarkResultsOverviewInput = {}) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/results${queryString({ limit: input.limit })}`, + ); + }, + + async getRunResults(benchmarkSlug, runId) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results`, + ); + }, + + async getRunTaskResults(benchmarkSlug, runId, input: BenchmarkRunTaskResultsInput = {}) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/tasks${queryString({ bucketSize: input.bucketSize, failureLimit: input.failureLimit })}`, + ); + }, + + async getRunTimeline(benchmarkSlug, runId, input: BenchmarkRunTimelineInput = {}) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/timeline${queryString({ bucketMs: input.bucketMs })}`, + ); + }, + + async getRunImports(benchmarkSlug, runId) { + return request( + 'GET', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/imports`, + ); + }, + + async submitRunSummary(benchmarkSlug, runId, input) { + await request( + 'POST', + `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/summary`, + input as unknown as JsonObject, + ); + }, + }; + + return client; +} + diff --git a/packages/benchsdk-api/src/index.ts b/packages/benchsdk-api/src/index.ts new file mode 100644 index 00000000..b52a3df8 --- /dev/null +++ b/packages/benchsdk-api/src/index.ts @@ -0,0 +1,74 @@ +export { BenchmarkApiError, createBenchmarkClient } from './client'; +export type { + BenchmarkAssignment, + BenchmarkArtifact, + BenchmarkClient, + BenchmarkClientConfig, + BenchmarkConcurrencyPoint, + BenchmarkAnalyticsReadiness, + BenchmarkEventRateBucket, + BenchmarkFailurePoint, + BenchmarkParticipant, + BenchmarkResource, + BenchmarkResultLatencySummary, + BenchmarkResultsOverview, + BenchmarkResultsOverviewAnalytics, + BenchmarkResultsOverviewInput, + BenchmarkResultsOverviewRun, + BenchmarkResultSummary, + BenchmarkRun, + BenchmarkRunImports, + BenchmarkRunAnalyticsSummary, + BenchmarkRunImportsSummary, + BenchmarkRunImportItem, + BenchmarkRunResults, + BenchmarkRunStatus, + BenchmarkRunSummaryInput, + BenchmarkRunSummaryMetric, + BenchmarkRunSummaryResult, + BenchmarkRunSummaryRunMetadata, + BenchmarkRunSummaryScalar, + BenchmarkRunTaskResults, + BenchmarkRunTaskResultsInput, + BenchmarkRunTimeline, + BenchmarkRunTimelineInput, + BenchmarkRunWorker, + BenchmarkStepResultSummary, + BenchmarkTaskBucket, + BenchmarkWorkerAttempt, + BenchmarkWorkerStatus, + ClaimWorkerInput, + CreateWorkerArtifactInput, + CreateWorkerArtifactResponse, + CreateRunInput, + DefineStepOptions, + JsonObject, + JsonValue, + PlanWorkersInput, + RunProgress, + RunProgressConcurrency, + RunProgressParticipant, + RunProgressParticipantCounts, + RunProgressStatus, + RunProgressSummary, + RunProgressTaskCounts, + RunProgressWorkerCounts, + RunWorkerContext, + RunWorkerOptions, + RunWorkerResult, + SendTaskResultsInput, + TaskStepRecord, + TaskResultRecord, + TaskResultsResponse, + TaskFunction, + UpdateBenchmarkInput, + UpdateParticipantInput, + UpdateRunInput, + UpdateWorkerInput, + UpsertBenchmarkInput, + UpsertParticipantInput, + UploadWorkerArtifactInput, + WorkerConcurrencySample, + WorkerFinishContext, + WorkerHeartbeatInput, +} from './types'; diff --git a/packages/benchsdk-api/src/types.ts b/packages/benchsdk-api/src/types.ts new file mode 100644 index 00000000..98a24716 --- /dev/null +++ b/packages/benchsdk-api/src/types.ts @@ -0,0 +1,745 @@ +export type JsonValue = string | number | boolean | null | JsonObject | JsonValue[]; +export type JsonObject = { [key: string]: JsonValue }; + +export interface BenchmarkClientConfig { + /** API base URL. Defaults to https://platform.computesdk.com/api/v1. */ + baseUrl?: string; + /** Bearer token. Defaults to process.env.COMPUTESDK_ADMIN_API_KEY, then process.env.COMPUTESDK_API_KEY. */ + apiKey?: string; + /** Custom fetch implementation, mostly useful for tests. */ + fetch?: typeof fetch; +} + +export interface BenchmarkResource { + id: string; + slug: string; + name: string; + status?: string; + config?: JsonObject; + defaultRunConfig?: JsonObject; +} + +export type BenchmarkRunStatus = 'planned' | 'in_progress' | 'completed' | 'failed'; +export type BenchmarkWorkerStatus = 'pending' | 'running' | 'completed' | 'failed'; + +export interface BenchmarkRun { + id: string; + benchmarkId: string; + name?: string | null; + status: BenchmarkRunStatus | string; + /** Idempotency key: runs created with the same key (per org + benchmark) are the same run. */ + runKey?: string | null; + totalTasks: number; + /** The run declared no size: `totalTasks` is the sum of what its participants declare. */ + participantSized?: boolean; + workerCount: number; + config?: JsonObject; + createdAt?: string; + updatedAt?: string; +} + +export interface BenchmarkParticipant { + id: string; + benchmarkId: string; + runId: string; + slug: string; + label?: string | null; + provider?: string | null; + status: BenchmarkRunStatus | string; + totalTasks: number; + workerCount: number; + config?: JsonObject; +} + +export interface BenchmarkRunWorker { + id: string; + benchmarkId: string; + runId: string; + participantId: string; + workerIndex: number; + workerCount: number; + taskIndexStart: number; + taskIndexEnd: number; + targetConcurrency: number; + status: BenchmarkWorkerStatus | string; + progressDone?: number; + progressInFlight?: number; + progressErrors?: number; + progressTotal?: number; + currentStep?: string | null; + concurrency?: WorkerConcurrencySample[]; +} + +export interface BenchmarkWorkerAttempt { + id: string; + benchmarkId: string; + runId: string; + participantId: string; + workerId: string; + attemptNumber: number; + status: string; +} + +export interface BenchmarkAssignment { + benchmarkId: string; + benchmarkSlug: string; + runId: string; + participantId: string; + participantSlug: string; + provider?: string | null; + workerId: string; + workerIndex: number; + workerCount: number; + attemptId: string; + attemptNumber: number; + taskRange: { + start: number; + end: number; + count: number; + }; + targetConcurrency: number; + config?: JsonObject; +} + +export interface UpsertBenchmarkInput { + name: string; + status?: string; + config?: JsonObject; + defaultRunConfig?: JsonObject; +} + +export interface UpdateBenchmarkInput { + name?: string; + status?: string; + config?: JsonObject; + defaultRunConfig?: JsonObject; +} + +export interface CreateRunInput { + /** + * Idempotency key for get-or-create: sibling callers passing the same key + * (per org + benchmark) converge on one run instead of each opening its own. + */ + runKey?: string; + /** Omit to open a participant-sized run: each participant declares its own size when it registers. */ + totalTasks?: number; + workerCount?: number; + participants?: string[]; + config?: JsonObject; +} + +export interface UpdateRunInput { + status?: BenchmarkRunStatus; + config?: JsonObject; +} + +export interface UpsertParticipantInput { + label?: string; + provider?: string; + status?: string; + totalTasks?: number; + workerCount?: number; + config?: JsonObject; +} + +export type UpdateParticipantInput = UpsertParticipantInput; + +export interface UpdateWorkerInput { + status?: BenchmarkWorkerStatus; + progressDone?: number; + progressInFlight?: number; + progressErrors?: number; + progressTotal?: number; +} + +export interface ClaimWorkerInput { + processKind?: string; + processKey?: string; +} + +export interface PlanWorkersInput { + workerCount?: number; + targetConcurrency?: number; + config?: JsonObject; +} + +export interface TaskResultRecord { + taskIndex: number; + status: string; + startedAt?: string; + completedAt?: string; + latencyMs?: number; + firstCommandMs?: number | null; + errorCode?: string | null; + steps?: TaskStepRecord[]; + data?: JsonObject; +} + +export interface TaskStepRecord { + name: string; + status: 'success' | 'error'; + startedAt?: string; + completedAt?: string; + latencyMs?: number; + errorCode?: string | null; + data?: JsonObject; + /** Number of parallel invocations requested for this step. */ + concurrency?: number; + /** Per-iteration timeout in milliseconds applied to this step. */ + timeoutMs?: number; +} + +export interface SendTaskResultsInput { + benchmarkSlug: string; + runId: string; + workerId: string; + attemptId: string; + sequenceNumber: number; + isFinal: boolean; + records: TaskResultRecord[]; +} + +export interface TaskResultsResponse { + accepted?: number; + eventBatchId?: string; + queued?: boolean; + eventBatch?: unknown; + duplicate?: boolean; + queueMessageId?: string; +} + +export interface CreateWorkerArtifactInput { + attemptId: string; + kind: string; + contentType?: string; + name?: string; + metadata?: JsonObject; +} + +export interface UploadWorkerArtifactInput extends CreateWorkerArtifactInput { + body: BodyInit; +} + +export interface BenchmarkArtifact { + id?: string; + artifactId?: string; + benchmarkId?: string; + runId?: string; + participantId?: string; + participantSlug?: string; + workerId?: string; + attemptId?: string; + kind: string; + name?: string | null; + contentType?: string | null; + objectKey?: string; + uploadUrl?: string; + uploadUrlExpiresAt?: string; + metadata?: JsonObject; + createdAt?: string; +} + +export interface CreateWorkerArtifactResponse { + artifact?: BenchmarkArtifact; + artifactId?: string; + uploadUrl?: string; + uploadUrlExpiresAt?: string; + objectKey?: string; +} + +export interface BenchmarkResultLatencySummary { + min: number | null; + avg: number | null; + p50: number | null; + p95: number | null; + p99: number | null; + max: number | null; +} + +export interface BenchmarkResultSummary { + taskCount: number; + successCount: number; + errorCount: number; + otherCount: number; + latencyCount: number; + successRate: number; + latencyMs: BenchmarkResultLatencySummary; + firstStartedAt: string | null; + lastCompletedAt: string | null; +} + +export interface BenchmarkParticipantResultSummary extends BenchmarkResultSummary { + participantSlug: string; + provider: string | null; +} + +export interface BenchmarkStepResultSummary { + participantSlug: string; + provider: string | null; + stepName: string; + stepCount: number; + successCount: number; + errorCount: number; + otherCount: number; + latencyCount: number; + successRate: number; + latencyMs: BenchmarkResultLatencySummary; +} + +export interface BenchmarkResultsOverviewInput { + limit?: number; +} + +export type BenchmarkAnalyticsReadiness = 'ready' | 'complete' | 'partial' | 'pending' | 'unavailable' | 'failed'; + +export interface BenchmarkRunAnalyticsSummary { + status: BenchmarkAnalyticsReadiness; + eventBatches: number; + persisted: number; + queued: number; + failed: number; + imports: { + pending: number; + importing: number; + imported: number; + failed: number; + missing: number; + }; +} + +export interface BenchmarkResultsOverviewAnalytics { + status: BenchmarkAnalyticsReadiness; + query: 'available' | 'unavailable'; + error?: string; +} + +export interface BenchmarkResultsOverviewRun { + run: BenchmarkRun; + analytics: BenchmarkRunAnalyticsSummary; + participants: Array; +} + +export interface BenchmarkResultsOverview { + benchmark: Pick; + generatedAt: string; + analytics: BenchmarkResultsOverviewAnalytics; + items: BenchmarkResultsOverviewRun[]; +} + +export interface BenchmarkRunResults { + benchmark: Pick; + run: Pick; + generatedAt: string; + overall: BenchmarkResultSummary; + participants: BenchmarkParticipantResultSummary[]; + steps: BenchmarkStepResultSummary[]; +} + +export interface BenchmarkRunTaskResultsInput { + bucketSize?: number; + failureLimit?: number; +} + +export interface BenchmarkTaskBucket { + participantSlug: string; + provider: string | null; + bucketStart: number; + bucketEnd: number; + taskIndexMidpoint: number; + taskCount: number; + successCount: number; + errorCount: number; + latencyMs: Pick; +} + +export interface BenchmarkFailurePoint { + participantSlug: string; + provider: string | null; + taskIndex: number; + errorCode: string | null; +} + +export interface BenchmarkRunTaskResults { + run: { id: string }; + generatedAt: string; + bucketSize: number; + buckets: BenchmarkTaskBucket[]; + failures: BenchmarkFailurePoint[]; +} + +export interface BenchmarkRunTimelineInput { + bucketMs?: number; +} + +export interface BenchmarkEventRateBucket { + participantSlug: string; + provider: string | null; + tMs: number; + completed: number; + succeeded: number; + failed: number; +} + +export interface BenchmarkConcurrencyPoint { + participantSlug: string; + provider: string | null; + workerId: string; + recordedAt: string; + tMs: number; + step: string; + active: number; + target: number; +} + +export interface BenchmarkRunTimeline { + run: { id: string }; + generatedAt: string; + eventRate: { + bucketMs: number; + buckets: BenchmarkEventRateBucket[]; + }; + concurrency: { + firstRecordedAt: string | null; + heartbeatCount: number; + points: BenchmarkConcurrencyPoint[]; + }; +} + +export interface BenchmarkRunImportsSummary { + eventBatches: number; + persisted: number; + queued: number; + failed: number; + imports: { + pending: number; + importing: number; + imported: number; + failed: number; + missing: number; + }; +} + +export interface BenchmarkRunImportItem { + eventBatchId: string; + batchType: string; + sequenceNumber: number; + batchStatus: string; + eventCount: number; + objectKey: string | null; + batchErrorMessage: string | null; + createdAt: string; + persistedAt: string | null; + sink: string | null; + importStatus: string | null; + importAttempts: number | null; + importedAt: string | null; + failedAt: string | null; + importErrorMessage: string | null; +} + +export interface BenchmarkRunImports { + run: { id: string }; + generatedAt: string; + summary: BenchmarkRunImportsSummary; + items: BenchmarkRunImportItem[]; +} + +export interface WorkerConcurrencySample { + step: string; + active: number; + target: number; +} + +export interface WorkerHeartbeatInput { + attemptId: string; + progressDone?: number; + progressInFlight?: number; + progressErrors?: number; + progressTotal?: number; + currentStep?: string | null; + concurrency?: WorkerConcurrencySample[]; +} + +export interface RunProgressConcurrency { + step: string; + active: number; + target: number; + ready: boolean; + freshWorkerCount: number; +} + +export type RunProgressStatus = 'planned' | 'in_progress' | 'completed' | 'failed'; + +export interface RunProgressWorkerCounts { + pending: number; + running: number; + completed: number; + failed: number; + stale: number; + total: number; +} + +export interface RunProgressTaskCounts { + done: number; + inFlight: number; + errors: number; + total: number; + completionRatio: number; +} + +export interface RunProgressParticipantCounts { + planned: number; + inProgress: number; + completed: number; + failed: number; + total: number; +} + +export interface RunProgressSummary { + status: RunProgressStatus; + started: boolean; + completed: boolean; + participants: RunProgressParticipantCounts; +} + +export interface RunProgressParticipant { + id: string; + slug: string; + provider?: string | null; + status: RunProgressStatus; + totalTasks: number; + workerCount: number; + workers: RunProgressWorkerCounts; + tasks: RunProgressTaskCounts; + concurrency: RunProgressConcurrency[]; +} + +export interface RunProgress { + run: { + id: string; + status: string; + totalTasks: number; + workerCount: number; + }; + summary: RunProgressSummary; + freshnessWindowSeconds: number; + generatedAt: string; + participants: RunProgressParticipant[]; +} + +export interface RunWorkerContext { + assignment: BenchmarkAssignment; + taskIndex: number; + step(name: string, fn: () => Promise | T, options?: DefineStepOptions): Promise; + /** + * Attaches a JSON measurement to the platform. Called inside a `step`, it + * lands on that step's `data`; called at task top-level, on the task record's + * `data`. Repeated calls merge (shallow). Use this for anything you want on + * the platform — step return values are control flow and are never recorded. + */ + measure(data: JsonObject): void; + /** Appends a line to the worker log, uploaded as an artifact when the worker finishes. */ + log(message: string, meta?: JsonObject): void; +} + +export interface WorkerFinishContext { + assignment: BenchmarkAssignment; + records: TaskResultRecord[]; + status: 'success' | 'error'; + client: BenchmarkClient; + uploadArtifact(input: Omit): Promise; +} + +export interface DefineStepOptions { + /** Report this step as active in heartbeat concurrency samples. Defaults to true. */ + reportConcurrency?: boolean; + /** Per-worker target for this step. Defaults to worker concurrency/assignment target. */ + concurrency?: number; + /** Number of parallel invocations the step function should run internally. Used by the runner to record step-level concurrency. */ + stepConcurrency?: number; + /** Per-invocation timeout in milliseconds for this step. Used by the runner to record step-level timeout metadata. */ + timeoutMs?: number; + /** Readiness coordination mode. Defaults to internal. */ + readiness?: 'poll' | 'internal'; + /** Poll interval while waiting for readiness. Defaults to 1000ms. */ + readyPollIntervalMs?: number; + /** Maximum time to wait for readiness. Defaults to no timeout. */ + readyTimeoutMs?: number; +} + +/** + * The unit of work a worker runs, once per task index. Steps are declared + * imperatively via `context.step(...)`; this is the sole task shape the worker + * engine accepts. Higher-level authoring (`defineTask`) lives in + * `@benchsdk/runner`, which compiles down to a function of this shape. + */ +export type TaskFunction = (context: RunWorkerContext) => Promise | JsonObject | void; + +export interface RunWorkerResult { + assignment: BenchmarkAssignment | null; + records: TaskResultRecord[]; +} + +export interface RunWorkerOptions { + benchmarkSlug: string; + runId: string; + participantSlug: string; + processKind?: string; + processKey?: string; + concurrency?: number; + batchSize?: number; + flushIntervalMs?: number; + heartbeatIntervalMs?: number; + readyPollIntervalMs?: number; + onResult?: (record: TaskResultRecord) => void; + /** Runs once after final result flush and before worker completion/failure is reported. */ + onFinish?: (context: WorkerFinishContext) => Promise | void; + task: TaskFunction; +} + +export interface BenchmarkRunSummaryMetric { + name: string; + unit: string; + median: number; + p95: number; + p99: number; +} + +export interface BenchmarkRunSummaryScalar { + name: string; + value: number; + unit: string; +} + +export interface BenchmarkRunSummaryResult { + provider: string; + dimensions?: Record; + metrics: BenchmarkRunSummaryMetric[]; + scalars?: BenchmarkRunSummaryScalar[]; + compositeScore: number; + successRate: number; + scoringVersion?: string | null; + skipped: boolean; + skipReason?: string | null; +} + +export interface BenchmarkRunSummaryRunMetadata { + gitSha?: string; + gitRef?: string; + triggeredBy?: string; + nodeVersion?: string; + platform?: string; + arch?: string; +} + +export interface BenchmarkRunSummaryInput { + run: BenchmarkRunSummaryRunMetadata; + results: BenchmarkRunSummaryResult[]; +} + +export interface BenchmarkClient { + upsertBenchmark(slug: string, input: UpsertBenchmarkInput): Promise; + updateBenchmark(slug: string, input: UpdateBenchmarkInput): Promise; + getBenchmark(slug: string): Promise; + listBenchmarks(): Promise; + createRun(benchmarkSlug: string, input: CreateRunInput): Promise<{ + run: BenchmarkRun; + participants: BenchmarkParticipant[]; + /** The slug of the org the run was attributed to, resolved server-side from the caller's API key. */ + organizationSlug: string; + }>; + listRuns(benchmarkSlug: string): Promise; + getRun(benchmarkSlug: string, runId: string): Promise; + updateRun(benchmarkSlug: string, runId: string, input: UpdateRunInput): Promise; + upsertParticipant( + benchmarkSlug: string, + runId: string, + participantSlug: string, + input?: UpsertParticipantInput, + ): Promise; + updateParticipant( + benchmarkSlug: string, + runId: string, + participantSlug: string, + input: UpdateParticipantInput, + ): Promise; + listParticipants(benchmarkSlug: string, runId: string): Promise; + getParticipant( + benchmarkSlug: string, + runId: string, + participantSlug: string, + ): Promise; + listWorkers( + benchmarkSlug: string, + runId: string, + participantSlug: string, + ): Promise; + planWorkers( + benchmarkSlug: string, + runId: string, + participantSlug: string, + input?: PlanWorkersInput, + ): Promise; + getWorker(benchmarkSlug: string, runId: string, workerId: string): Promise; + updateWorker( + benchmarkSlug: string, + runId: string, + workerId: string, + input: UpdateWorkerInput, + ): Promise; + getRunProgress(benchmarkSlug: string, runId: string): Promise; + claimWorker( + benchmarkSlug: string, + runId: string, + participantSlug: string, + input?: ClaimWorkerInput, + ): Promise; + releaseWorker(benchmarkSlug: string, runId: string, workerId: string, attemptId: string): Promise<{ + worker: BenchmarkRunWorker; + attempt: BenchmarkWorkerAttempt; + }>; + sendTaskResults(input: SendTaskResultsInput): Promise; + heartbeatWorker(benchmarkSlug: string, runId: string, workerId: string, input: WorkerHeartbeatInput): Promise<{ + worker: BenchmarkRunWorker; + attempt: BenchmarkWorkerAttempt; + }>; + completeWorker(benchmarkSlug: string, runId: string, workerId: string, attemptId: string): Promise<{ + worker: BenchmarkRunWorker; + attempt: BenchmarkWorkerAttempt; + }>; + failWorker( + benchmarkSlug: string, + runId: string, + workerId: string, + attemptId: string, + error?: unknown, + ): Promise<{ worker: BenchmarkRunWorker; attempt: BenchmarkWorkerAttempt }>; + createWorkerArtifact( + benchmarkSlug: string, + runId: string, + workerId: string, + input: CreateWorkerArtifactInput, + ): Promise; + uploadWorkerArtifact( + benchmarkSlug: string, + runId: string, + workerId: string, + input: UploadWorkerArtifactInput, + ): Promise; + listRunArtifacts(benchmarkSlug: string, runId: string): Promise; + listWorkerArtifacts(benchmarkSlug: string, runId: string, workerId: string): Promise; + getBenchmarkResults(benchmarkSlug: string, input?: BenchmarkResultsOverviewInput): Promise; + getRunResults(benchmarkSlug: string, runId: string): Promise; + getRunTaskResults( + benchmarkSlug: string, + runId: string, + input?: BenchmarkRunTaskResultsInput, + ): Promise; + getRunTimeline( + benchmarkSlug: string, + runId: string, + input?: BenchmarkRunTimelineInput, + ): Promise; + getRunImports(benchmarkSlug: string, runId: string): Promise; + submitRunSummary(benchmarkSlug: string, runId: string, input: BenchmarkRunSummaryInput): Promise; +} diff --git a/packages/benchsdk-api/tsconfig.json b/packages/benchsdk-api/tsconfig.json new file mode 100644 index 00000000..ed464a96 --- /dev/null +++ b/packages/benchsdk-api/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/benchsdk-api/tsup.config.ts b/packages/benchsdk-api/tsup.config.ts new file mode 100644 index 00000000..d1eea6f6 --- /dev/null +++ b/packages/benchsdk-api/tsup.config.ts @@ -0,0 +1,16 @@ +import { defineConfig } from 'tsup'; +import pkg from './package.json'; + +export default defineConfig({ + entry: { + index: 'src/index.ts', + }, + format: ['cjs', 'esm'], + dts: true, + splitting: false, + sourcemap: true, + clean: true, + define: { + __BENCH_VERSION__: JSON.stringify(pkg.version), + }, +}); diff --git a/packages/benchsdk-api/vitest.config.ts b/packages/benchsdk-api/vitest.config.ts new file mode 100644 index 00000000..014f97ef --- /dev/null +++ b/packages/benchsdk-api/vitest.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { + environment: 'node', + globals: true, + }, +}); diff --git a/packages/benchsdk-worker/package.json b/packages/benchsdk-worker/package.json new file mode 100644 index 00000000..72493e6c --- /dev/null +++ b/packages/benchsdk-worker/package.json @@ -0,0 +1,65 @@ +{ + "name": "@benchsdk/worker", + "version": "0.1.0", + "private": false, + "type": "module", + "description": "Benchmark worker runtime: claim an assignment, run tasks, flush results and heartbeats", + "author": "Garrison", + "license": "MIT", + "main": "./dist/index.cjs", + "module": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js", + "require": "./dist/index.cjs" + } + }, + "files": [ + "dist", + "README.md" + ], + "scripts": { + "build": "tsup", + "clean": "rimraf dist", + "dev": "tsup --watch", + "lint": "eslint", + "prepare": "tsup", + "pretest": "tsup", + "test": "vitest run", + "test:coverage": "vitest run --coverage", + "test:watch": "vitest watch", + "typecheck": "tsc --noEmit" + }, + "keywords": [ + "computesdk", + "benchmark", + "worker", + "orchestrator" + ], + "repository": { + "type": "git", + "url": "https://github.com/computesdk/benchmarks.git", + "directory": "packages/benchsdk-worker" + }, + "homepage": "https://www.computesdk.com", + "bugs": { + "url": "https://github.com/computesdk/benchmarks/issues" + }, + "engines": { + "node": ">=18.0.0" + }, + "dependencies": { + "@benchsdk/api": "workspace:*" + }, + "devDependencies": { + "@types/node": "^20.0.0", + "@vitest/coverage-v8": "^1.0.0", + "eslint": "^8.37.0", + "rimraf": "^5.0.0", + "tsup": "^8.0.0", + "typescript": "^5.0.0", + "vitest": "^1.0.0" + } +} diff --git a/packages/benchsdk-worker/src/index.ts b/packages/benchsdk-worker/src/index.ts new file mode 100644 index 00000000..7c3b0c31 --- /dev/null +++ b/packages/benchsdk-worker/src/index.ts @@ -0,0 +1,19 @@ +export { runWorker } from './worker'; +export { BenchmarkReporter, claimBenchmarkReporter } from './reporter'; +export { createSystemMetricsCollector } from './metrics'; +export { filterParticipantsByEnv, selectParticipants } from './participants'; +export type { + BenchmarkReporterArtifactInput, + BenchmarkReporterBarrierInput, + BenchmarkReporterBarrierResult, + BenchmarkReporterConfig, + BenchmarkReporterHeartbeatInput, + BenchmarkReporterProgress, +} from './reporter'; +export type { + BenchmarkSystemMetricsCollector, + BenchmarkSystemMetricsSample, +} from './metrics'; +export type { + BaseParticipant, +} from './participants'; diff --git a/packages/benchsdk-worker/src/metrics.ts b/packages/benchsdk-worker/src/metrics.ts new file mode 100644 index 00000000..77dcec14 --- /dev/null +++ b/packages/benchsdk-worker/src/metrics.ts @@ -0,0 +1,93 @@ +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import { monitorEventLoopDelay } from 'node:perf_hooks'; + +export interface BenchmarkSystemMetricsSample { + ts: string; + uptimeMs: number; + cpuUserUs: number; + cpuSystemUs: number; + memRssMb: number; + memHeapUsedMb: number; + memHeapTotalMb: number; + memExternalMb: number; + eventLoopP50Ms: number; + eventLoopP99Ms: number; + eventLoopMaxMs: number; + loadavg1m: number; + loadavg5m: number; + loadavg15m: number; + openFds: number | null; + sockstat: Record | null; +} + +export interface BenchmarkSystemMetricsCollector { + sample(): BenchmarkSystemMetricsSample; + stop(): void; +} + +function readSockstat(): Record | null { + try { + const data = fs.readFileSync('/proc/net/sockstat', 'utf-8'); + const out: Record = {}; + for (const line of data.split('\n')) { + const index = line.indexOf(':'); + if (index < 0) continue; + const section = line.slice(0, index).trim().toLowerCase(); + const parts = line.slice(index + 1).trim().split(/\s+/); + for (let i = 0; i + 1 < parts.length; i += 2) { + const value = Number.parseInt(parts[i + 1], 10); + if (!Number.isNaN(value)) out[`${section}_${parts[i]}`] = value; + } + } + return out; + } catch { + return null; + } +} + +function countOpenFds(): number | null { + try { + return fs.readdirSync('/proc/self/fd').length; + } catch { + return null; + } +} + +export function createSystemMetricsCollector(): BenchmarkSystemMetricsCollector { + const startedAt = Date.now(); + const cpuBaseline = process.cpuUsage(); + const eventLoop = monitorEventLoopDelay({ resolution: 20 }); + eventLoop.enable(); + + return { + sample() { + const cpu = process.cpuUsage(cpuBaseline); + const memory = process.memoryUsage(); + const loadavg = os.loadavg(); + const sample: BenchmarkSystemMetricsSample = { + ts: new Date().toISOString(), + uptimeMs: Date.now() - startedAt, + cpuUserUs: cpu.user, + cpuSystemUs: cpu.system, + memRssMb: Math.round(memory.rss / 1024 / 1024), + memHeapUsedMb: Math.round(memory.heapUsed / 1024 / 1024), + memHeapTotalMb: Math.round(memory.heapTotal / 1024 / 1024), + memExternalMb: Math.round(memory.external / 1024 / 1024), + eventLoopP50Ms: eventLoop.percentile(50) / 1e6, + eventLoopP99Ms: eventLoop.percentile(99) / 1e6, + eventLoopMaxMs: eventLoop.max / 1e6, + loadavg1m: loadavg[0], + loadavg5m: loadavg[1], + loadavg15m: loadavg[2], + openFds: countOpenFds(), + sockstat: readSockstat(), + }; + eventLoop.reset(); + return sample; + }, + stop() { + eventLoop.disable(); + }, + }; +} diff --git a/packages/benchsdk-worker/src/participants.ts b/packages/benchsdk-worker/src/participants.ts new file mode 100644 index 00000000..806acfa8 --- /dev/null +++ b/packages/benchsdk-worker/src/participants.ts @@ -0,0 +1,54 @@ +/** + * Base interface for a benchmark participant — the shared shape across all + * benchmark categories (sandbox, ai-gateway, browser, storage). Each category + * extends this with its own provider-specific fields. + */ +export interface BaseParticipant { + /** Participant name (e.g. 'e2b', 'daytona', 'openrouter') */ + name: string; + /** Environment variables that must all be set to run this participant */ + requiredEnvVars: string[]; +} + +/** +/** + * Filters `participants` down to those whose `requiredEnvVars` are all set + * in `process.env`. Returns an object `{ available, skipped }` where `skipped` + * includes the names and missing vars for logging. + */ +export function filterParticipantsByEnv( + participants: T[], +): { available: T[]; skipped: { name: string; missing: string[] }[] } { + const available: T[] = []; + const skipped: { name: string; missing: string[] }[] = []; + + for (const p of participants) { + const missing = p.requiredEnvVars.filter((v) => !process.env[v]); + if (missing.length > 0) { + skipped.push({ name: p.name, missing }); + } else { + available.push(p); + } + } + + return { available, skipped }; +} + +/** + * Filters `all` down to the requested `names`, exiting with a clear error + * if any name is unrecognized. Returns `all` unchanged when `names` is + * undefined (no filter specified). + */ +export function selectParticipants( + all: T[], + names?: string[], +): T[] { + if (!names) return all; + const unknown = names.filter((n) => !all.some((p) => p.name === n)); + if (unknown.length > 0) { + console.error(`Unknown participant(s): ${unknown.join(', ')}`); + console.error(`Available: ${all.map((p) => p.name).join(', ')}`); + process.exit(1); + } + return all.filter((p) => names.includes(p.name)); +} diff --git a/packages/benchsdk-worker/src/reporter.ts b/packages/benchsdk-worker/src/reporter.ts new file mode 100644 index 00000000..34048b7d --- /dev/null +++ b/packages/benchsdk-worker/src/reporter.ts @@ -0,0 +1,238 @@ +import { createBenchmarkClient } from '@benchsdk/api'; +import type { + BenchmarkAssignment, + BenchmarkClient, + BenchmarkClientConfig, + CreateWorkerArtifactResponse, + JsonObject, + TaskResultRecord, + WorkerConcurrencySample, +} from '@benchsdk/api'; + +const DEFAULT_REPORTER_BATCH_SIZE = 500; +const DEFAULT_READY_POLL_INTERVAL_MS = 1000; + +export interface BenchmarkReporterConfig extends BenchmarkClientConfig { + benchmarkSlug: string; + runId: string; + participantSlug: string; + processKind?: string; + processKey?: string; + batchSize?: number; +} + +export interface BenchmarkReporterProgress { + done: number; + inFlight: number; + errors: number; + total?: number; +} + +export interface BenchmarkReporterArtifactInput { + kind: string; + name?: string; + contentType?: string; + body: BodyInit; + metadata?: JsonObject; +} + +export interface BenchmarkReporterHeartbeatInput { + currentStep?: string | null; + concurrency?: WorkerConcurrencySample[]; +} + +export interface BenchmarkReporterBarrierInput { + step: string; + timeoutMs?: number; + pollIntervalMs?: number; + active?: number; + target?: number; + concurrency?: WorkerConcurrencySample[]; +} + +export interface BenchmarkReporterBarrierResult { + active: number | null; + target: number | null; + ready: boolean; + measuredAt: string; +} + +export class BenchmarkReporter { + private readonly client: BenchmarkClient; + private readonly assignment: BenchmarkAssignment; + private readonly cfg: Required>; + private pending: TaskResultRecord[] = []; + private sequenceNumber = 0; + private flushChain: Promise = Promise.resolve(); + private progress: BenchmarkReporterProgress; + private barrier: { step: string; concurrency: WorkerConcurrencySample[] } | null = null; + + private constructor(client: BenchmarkClient, cfg: BenchmarkReporterConfig, assignment: BenchmarkAssignment) { + this.client = client; + this.assignment = assignment; + this.cfg = { + benchmarkSlug: cfg.benchmarkSlug, + runId: cfg.runId, + participantSlug: cfg.participantSlug, + batchSize: cfg.batchSize ?? DEFAULT_REPORTER_BATCH_SIZE, + }; + this.progress = { done: 0, inFlight: 0, errors: 0, total: assignment.taskRange.count }; + } + + static async claim(cfg: BenchmarkReporterConfig): Promise { + const client = createBenchmarkClient(cfg); + try { + const assignment = await client.claimWorker(cfg.benchmarkSlug, cfg.runId, cfg.participantSlug, { + processKind: cfg.processKind, + processKey: cfg.processKey, + }); + return assignment ? new BenchmarkReporter(client, cfg, assignment) : null; + } catch (error) { + console.warn( + `[benchsdk] failed to claim worker for ${cfg.benchmarkSlug}/${cfg.participantSlug}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + get workerAssignment(): BenchmarkAssignment { + return this.assignment; + } + + get taskCount(): number { + return this.assignment.taskRange.count; + } + + get taskIndexStart(): number { + return this.assignment.taskRange.start; + } + + setProgress(progress: BenchmarkReporterProgress): void { + this.progress = { ...progress, total: progress.total ?? this.assignment.taskRange.count }; + } + + recordResult(record: TaskResultRecord): void { + this.pending.push(record); + if (this.pending.length >= this.cfg.batchSize) void this.flush(false); + } + + async heartbeat(input: BenchmarkReporterHeartbeatInput = {}): Promise { + const barrier = this.barrier; + const currentStep = barrier?.step ?? input.currentStep; + const concurrency = barrier?.concurrency ?? input.concurrency; + await this.client.heartbeatWorker(this.cfg.benchmarkSlug, this.cfg.runId, this.assignment.workerId, { + attemptId: this.assignment.attemptId, + progressDone: this.progress.done, + progressInFlight: this.progress.inFlight, + progressErrors: this.progress.errors, + progressTotal: this.progress.total, + ...(currentStep ? { currentStep } : {}), + ...(concurrency ? { concurrency } : {}), + }).catch(() => {}); + } + + async waitForStepReady(input: BenchmarkReporterBarrierInput): Promise { + const startedAt = Date.now(); + const pollIntervalMs = input.pollIntervalMs ?? DEFAULT_READY_POLL_INTERVAL_MS; + const concurrency = input.concurrency ?? [{ + step: input.step, + active: input.active ?? this.assignment.taskRange.count, + target: input.target ?? this.assignment.taskRange.count, + }]; + this.barrier = { step: input.step, concurrency }; + try { + while (true) { + await this.heartbeat({ currentStep: input.step, concurrency }); + const progress = await this.client.getRunProgress(this.cfg.benchmarkSlug, this.cfg.runId).catch(() => null); + const participant = progress?.participants.find((item) => item.slug === this.cfg.participantSlug); + const step = participant?.concurrency.find((item) => item.step === input.step); + if (step?.ready) { + return { + active: step.active, + target: step.target, + ready: true, + measuredAt: progress?.generatedAt ?? new Date().toISOString(), + }; + } + if (input.timeoutMs !== undefined && Date.now() - startedAt >= input.timeoutMs) { + throw new Error(`Timed out waiting for benchmark step "${input.step}" to become ready.`); + } + await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); + } + } finally { + this.barrier = null; + } + } + + uploadArtifact(input: BenchmarkReporterArtifactInput): Promise { + return this.client.uploadWorkerArtifact(this.cfg.benchmarkSlug, this.cfg.runId, this.assignment.workerId, { + attemptId: this.assignment.attemptId, + kind: input.kind, + name: input.name, + contentType: input.contentType, + metadata: input.metadata, + body: input.body, + }).catch((error) => { + console.warn( + `[benchsdk] failed to upload ${input.kind} artifact for worker ${this.assignment.workerId}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + return null; + }); + } + + flush(isFinal = false): Promise { + this.flushChain = this.flushChain.then(async () => { + while (this.pending.length >= this.cfg.batchSize || (isFinal && this.pending.length > 0)) { + const batch = this.pending.slice(0, this.cfg.batchSize); + try { + await this.client.sendTaskResults({ + benchmarkSlug: this.cfg.benchmarkSlug, + runId: this.cfg.runId, + workerId: this.assignment.workerId, + attemptId: this.assignment.attemptId, + sequenceNumber: this.sequenceNumber, + isFinal: isFinal && batch.length === this.pending.length, + records: batch, + }); + } catch (error) { + // Results that never reach the platform are invisible otherwise: the + // worker still completes and the run just reports fewer tasks. + console.warn( + `[benchsdk] dropping ${this.pending.length} unsent task result(s) for worker ${this.assignment.workerId}: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + break; + } + this.pending.splice(0, batch.length); + this.sequenceNumber += 1; + } + }); + return this.flushChain; + } + + async finish(failed = false, error?: unknown): Promise { + await this.flush(true).catch(() => {}); + if (failed) { + await this.client.failWorker( + this.cfg.benchmarkSlug, + this.cfg.runId, + this.assignment.workerId, + this.assignment.attemptId, + error, + ).catch(() => {}); + return; + } + await this.client.completeWorker( + this.cfg.benchmarkSlug, + this.cfg.runId, + this.assignment.workerId, + this.assignment.attemptId, + ).catch(() => {}); + } +} + +export function claimBenchmarkReporter(config: BenchmarkReporterConfig): Promise { + return BenchmarkReporter.claim(config); +} diff --git a/packages/benchsdk-worker/src/worker.ts b/packages/benchsdk-worker/src/worker.ts new file mode 100644 index 00000000..154076a6 --- /dev/null +++ b/packages/benchsdk-worker/src/worker.ts @@ -0,0 +1,396 @@ +import type { + BenchmarkClient, + BenchmarkAssignment, + DefineStepOptions, + JsonObject, + RunWorkerContext, + RunWorkerOptions, + RunWorkerResult, + TaskFunction, + TaskResultRecord, + TaskStepRecord, + WorkerConcurrencySample, +} from '@benchsdk/api'; + +const DEFAULT_BATCH_SIZE = 1000; +const DEFAULT_HEARTBEAT_INTERVAL_MS = 30_000; +const DEFAULT_FLUSH_INTERVAL_MS = 30_000; +const DEFAULT_READY_POLL_INTERVAL_MS = 1000; +const MAX_TASK_RESULT_RECORDS = 5000; +const MAX_TASK_RECORD_STEPS = 100; +const MAX_HEARTBEAT_CONCURRENCY_SAMPLES = 20; +const MAX_WORKER_LOG_LINES = 100_000; + +function getErrorCode(error: unknown): string { + if (error instanceof Error && 'code' in error && typeof (error as { code: unknown }).code === 'string' && (error as { code: string }).code) { + return (error as { code: string }).code; + } + if (error instanceof Error && error.name) return error.name; + return 'ERROR'; +} + +function toJsonObject(value: unknown): JsonObject | undefined { + if (!value || typeof value !== 'object' || Array.isArray(value)) return undefined; + return value as JsonObject; +} + +function mergeMeasures(measures: JsonObject, returned: JsonObject | undefined): JsonObject | undefined { + const merged = { ...measures, ...(returned ?? {}) }; + return Object.keys(merged).length > 0 ? merged : undefined; +} + +function implicitTaskStep(record: TaskResultRecord, measures: JsonObject): TaskStepRecord { + return { + name: 'task', + status: record.status === 'success' ? 'success' : 'error', + startedAt: record.startedAt, + completedAt: record.completedAt, + latencyMs: record.latencyMs, + errorCode: record.errorCode ?? null, + data: Object.keys(measures).length > 0 ? { ...measures } : undefined, + }; +} + +function validatePositiveInteger(name: string, value: number): void { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`Benchmark ${name} must be a positive integer.`); + } +} + +function validateBatchSize(value: number): void { + validatePositiveInteger('batchSize', value); + if (value > MAX_TASK_RESULT_RECORDS) { + throw new Error(`Benchmark batchSize must be at most ${MAX_TASK_RESULT_RECORDS}.`); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function mapPool(items: T[], concurrency: number, fn: (item: T) => Promise): Promise { + let nextIndex = 0; + const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => { + while (nextIndex < items.length) { + const item = items[nextIndex]; + nextIndex += 1; + await fn(item); + } + }); + await Promise.all(workers); +} + +export async function runWorker(client: BenchmarkClient, options: RunWorkerOptions): Promise { + if (options.concurrency !== undefined) validatePositiveInteger('concurrency', options.concurrency); + if (options.batchSize !== undefined) validateBatchSize(options.batchSize); + if (options.flushIntervalMs !== undefined) validatePositiveInteger('flushIntervalMs', options.flushIntervalMs); + + const assignment = await client.claimWorker(options.benchmarkSlug, options.runId, options.participantSlug, { + processKind: options.processKind, + processKey: options.processKey, + }); + if (!assignment) return { assignment: null, records: [] }; + const claimed = assignment; + + let sequenceNumber = 0; + const records: TaskResultRecord[] = []; + const pending: TaskResultRecord[] = []; + const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; + const workerConcurrency = options.concurrency ?? claimed.targetConcurrency; + validatePositiveInteger('concurrency', workerConcurrency); + const taskIndices = Array.from({ length: claimed.taskRange.count }, (_, index) => claimed.taskRange.start + index); + const activeByStep = new Map(); + const targetByStep = new Map(); + const readyWaitByStep = new Map>(); + let doneCount = 0; + let errorCount = 0; + let inFlightCount = 0; + let flushChain = Promise.resolve(); + + function concurrencySamples(): WorkerConcurrencySample[] { + return Array.from(activeByStep.entries()) + .filter(([, active]) => active > 0) + .map(([step, active]) => ({ + step, + active, + target: targetByStep.get(step) ?? workerConcurrency, + })) + .sort((a, b) => b.active - a.active) + .slice(0, MAX_HEARTBEAT_CONCURRENCY_SAMPLES); + } + + async function sendHeartbeat(): Promise { + const concurrency = concurrencySamples(); + const step = concurrency[0]?.step ?? null; + await client.heartbeatWorker(options.benchmarkSlug, options.runId, claimed.workerId, { + attemptId: claimed.attemptId, + progressDone: doneCount, + progressInFlight: inFlightCount, + progressErrors: errorCount, + progressTotal: taskIndices.length, + ...(step ? { currentStep: step } : {}), + concurrency, + }); + } + + let heartbeatInFlight: Promise | null = null; + let heartbeatRequested = false; + + function requestHeartbeat(): void { + heartbeatRequested = true; + if (heartbeatInFlight) return; + + heartbeatInFlight = (async () => { + while (heartbeatRequested) { + heartbeatRequested = false; + await sendHeartbeat().catch(() => {}); + } + })().finally(() => { + heartbeatInFlight = null; + if (heartbeatRequested) requestHeartbeat(); + }); + } + + async function pollStepReady(stepName: string, stepOptions: DefineStepOptions): Promise { + const startedAt = Date.now(); + const pollInterval = stepOptions.readyPollIntervalMs ?? options.readyPollIntervalMs ?? DEFAULT_READY_POLL_INTERVAL_MS; + + while (true) { + const progress = await client.getRunProgress(options.benchmarkSlug, options.runId); + const participant = progress.participants.find((item) => item.slug === options.participantSlug); + const step = participant?.concurrency.find((item) => item.step === stepName); + if (step?.ready) return; + + if (typeof stepOptions.readyTimeoutMs === 'number' && Date.now() - startedAt >= stepOptions.readyTimeoutMs) { + throw new Error(`Timed out waiting for benchmark step "${stepName}" to become ready.`); + } + + await sleep(pollInterval); + } + } + + async function waitForStepReady(stepName: string, stepOptions: DefineStepOptions): Promise { + const existing = readyWaitByStep.get(stepName); + if (existing) return existing; + + const wait = pollStepReady(stepName, stepOptions).finally(() => { + if (readyWaitByStep.get(stepName) === wait) readyWaitByStep.delete(stepName); + }); + readyWaitByStep.set(stepName, wait); + return wait; + } + + const heartbeat = setInterval(() => { + requestHeartbeat(); + }, options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS); + heartbeat.unref?.(); + + async function flush(isFinal: boolean, force = false): Promise { + flushChain = flushChain.then(async () => { + if (force && doneCount >= taskIndices.length) return; + while (pending.length >= batchSize || ((isFinal || force) && pending.length > 0)) { + const batch = pending.splice(0, batchSize); + await client.sendTaskResults({ + benchmarkSlug: options.benchmarkSlug, + runId: options.runId, + workerId: claimed.workerId, + attemptId: claimed.attemptId, + sequenceNumber, + isFinal: isFinal && pending.length === 0, + records: batch, + }); + sequenceNumber += 1; + } + }); + await flushChain; + } + + const resultFlush = setInterval(() => { + if (doneCount < taskIndices.length) void flush(false, true).catch(() => {}); + }, options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS); + resultFlush.unref?.(); + + // Accumulated across the worker's tasks via `ctx.step` and `ctx.log`, + // uploaded once as a worker log artifact when the worker finishes. + const workerLogLines: string[] = []; + let workerLogTruncated = false; + function appendWorkerLog(line: string): void { + if (workerLogLines.length >= MAX_WORKER_LOG_LINES) { + if (!workerLogTruncated) { + workerLogTruncated = true; + workerLogLines.push('... (worker log truncated)'); + } + return; + } + workerLogLines.push(line); + } + + async function runFinishHook(status: 'success' | 'error'): Promise { + await options.onFinish?.({ + assignment: claimed, + records, + status, + client, + uploadArtifact(input) { + return client.uploadWorkerArtifact(options.benchmarkSlug, options.runId, claimed.workerId, { + ...input, + attemptId: claimed.attemptId, + }); + }, + }); + } + + async function uploadWorkerLogArtifact(): Promise { + if (workerLogLines.length === 0) return; + try { + await client.uploadWorkerArtifact(options.benchmarkSlug, options.runId, claimed.workerId, { + attemptId: claimed.attemptId, + kind: 'coordinator.log', + contentType: 'text/plain', + name: 'worker.log', + body: workerLogLines.join('\n') + '\n', + }); + } catch { + // Log upload is best-effort; never fail the run over it. + } + } + + try { + await sendHeartbeat().catch(() => {}); + + await mapPool(taskIndices, workerConcurrency, async (taskIndex) => { + inFlightCount += 1; + const startedAtDate = new Date(); + const startedAtMs = Date.now(); + const record: TaskResultRecord = { + taskIndex, + status: 'success', + startedAt: startedAtDate.toISOString(), + }; + const steps: TaskStepRecord[] = []; + const taskMeasures: JsonObject = {}; + // The step a `measure(...)` call currently attributes to. Set while a + // step's fn runs; null at task top-level (measures go on the record). + let activeStep: TaskStepRecord | null = null; + + function measure(data: JsonObject): void { + if (activeStep) { + activeStep.data = { ...(activeStep.data ?? {}), ...data }; + } else { + Object.assign(taskMeasures, data); + } + } + + function log(message: string, meta?: JsonObject): void { + const suffix = meta && Object.keys(meta).length > 0 ? ` ${JSON.stringify(meta)}` : ''; + appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${message}${suffix}`); + } + + async function step(name: string, fn: () => Promise | T, stepOptions: DefineStepOptions = {}): Promise { + const stepStartedAtMs = Date.now(); + const stepRecord: TaskStepRecord = { + name, + status: 'success', + startedAt: new Date().toISOString(), + completedAt: new Date().toISOString(), + latencyMs: 0, + }; + if (stepOptions.timeoutMs !== undefined) stepRecord.timeoutMs = stepOptions.timeoutMs; + if (stepOptions.stepConcurrency !== undefined) stepRecord.concurrency = stepOptions.stepConcurrency; + + const shouldReportConcurrency = stepOptions.reportConcurrency ?? true; + if (shouldReportConcurrency) { + const stepConcurrency = stepOptions.concurrency ?? workerConcurrency; + validatePositiveInteger(`step "${name}" concurrency`, stepConcurrency); + targetByStep.set(name, stepConcurrency); + activeByStep.set(name, (activeByStep.get(name) ?? 0) + 1); + requestHeartbeat(); + } + + const previousStep = activeStep; + activeStep = stepRecord; + try { + if (stepOptions.readiness === 'poll') { + await waitForStepReady(name, stepOptions); + } + const value = await fn(); + appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${name}`); + return value; + } catch (error) { + stepRecord.status = 'error'; + stepRecord.errorCode = getErrorCode(error); + appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${name}`); + appendWorkerLog(` error: ${error instanceof Error ? error.message : String(error)}`); + throw error; + } finally { + activeStep = previousStep; + stepRecord.completedAt = new Date().toISOString(); + stepRecord.latencyMs = Date.now() - stepStartedAtMs; + steps.push(stepRecord); + if (shouldReportConcurrency) { + const nextActive = Math.max(0, (activeByStep.get(name) ?? 0) - 1); + if (nextActive === 0) { + activeByStep.delete(name); + targetByStep.delete(name); + } else { + activeByStep.set(name, nextActive); + } + requestHeartbeat(); + } + } + } + + try { + const data = await options.task({ assignment: claimed, taskIndex, step, measure, log }); + record.data = mergeMeasures(taskMeasures, toJsonObject(data)); + } catch (error) { + record.status = 'error'; + record.errorCode = getErrorCode(error); + record.data = mergeMeasures(taskMeasures, { errorMessage: error instanceof Error ? error.message : String(error) }); + } finally { + record.completedAt = new Date().toISOString(); + record.latencyMs = Date.now() - startedAtMs; + // A task with no explicit steps is recorded as a single implicit + // 'task' step, so every task contributes at least one step row. + if (steps.length === 0) { + steps.push(implicitTaskStep(record, taskMeasures)); + } + record.steps = steps; + doneCount += 1; + inFlightCount = Math.max(0, inFlightCount - 1); + if (record.status !== 'success') errorCount += 1; + } + + records.push(record); + pending.push(record); + options.onResult?.(record); + if (pending.length >= batchSize) await flush(false); + }); + + await flush(true); + + const hasErrors = records.some((record) => record.status !== 'success'); + try { + await runFinishHook(hasErrors ? 'error' : 'success'); + } catch (error) { + if (!hasErrors) throw error; + } + + if (hasErrors) { + await client.failWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId, new Error('One or more tasks failed')); + } else { + await client.completeWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId); + } + + return { assignment: claimed, records }; + } catch (error) { + await flush(true).catch(() => {}); + await runFinishHook('error').catch(() => {}); + await client.failWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId, error).catch(() => {}); + throw error; + } finally { + await uploadWorkerLogArtifact(); + clearInterval(heartbeat); + clearInterval(resultFlush); + } +} diff --git a/packages/benchsdk-worker/tsconfig.json b/packages/benchsdk-worker/tsconfig.json new file mode 100644 index 00000000..ed464a96 --- /dev/null +++ b/packages/benchsdk-worker/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/benchsdk-worker/tsup.config.ts b/packages/benchsdk-worker/tsup.config.ts new file mode 100644 index 00000000..c4a23c0b --- /dev/null +++ b/packages/benchsdk-worker/tsup.config.ts @@ -0,0 +1,13 @@ +import { defineConfig } from 'tsup'; + +export default defineConfig({ + entry: { + index: 'src/index.ts', + }, + format: ['cjs', 'esm'], + dts: true, + splitting: false, + sourcemap: true, + clean: true, + external: ['@benchsdk/api'], +}); diff --git a/packages/benchsdk-worker/vitest.config.ts b/packages/benchsdk-worker/vitest.config.ts new file mode 100644 index 00000000..014f97ef --- /dev/null +++ b/packages/benchsdk-worker/vitest.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { + environment: 'node', + globals: true, + }, +}); diff --git a/packages/benchsdk/package.json b/packages/benchsdk/package.json index cf427e85..21ca7ace 100644 --- a/packages/benchsdk/package.json +++ b/packages/benchsdk/package.json @@ -51,7 +51,10 @@ "engines": { "node": ">=18.0.0" }, - "dependencies": {}, + "dependencies": { + "@benchsdk/api": "workspace:*", + "@benchsdk/worker": "workspace:*" + }, "devDependencies": { "@types/node": "^20.0.0", "@vitest/coverage-v8": "^1.0.0", diff --git a/packages/benchsdk/src/__tests__/public-api.contract.test.ts b/packages/benchsdk/src/__tests__/public-api.contract.test.ts index 892f7dd6..155f503f 100644 --- a/packages/benchsdk/src/__tests__/public-api.contract.test.ts +++ b/packages/benchsdk/src/__tests__/public-api.contract.test.ts @@ -1421,25 +1421,47 @@ describe('public API surface and type exports', () => { it('VAL-SDK-090: built barrel re-exports exactly the public type set and excludes internal types', () => { // Pin the actual shipped surface by reading the built dist/index.d.ts - // declarations rather than regex-parsing the source barrel. This catches - // drift between the source export list and what consumers receive. + // declarations. The split into @benchsdk/api and @benchsdk/worker means the + // barrel can contain both local type declarations and re-exports from workspace + // packages, so we gather all exported type names across the file. const distDecl = readFileSync(join(here, '..', '..', 'dist', 'index.d.ts'), 'utf8'); - // The built barrel emits a single `export { ... }` statement. Type-only - // re-exports are prefixed with `type `; parse those to recover the exact - // exported type set and compare it against the expected surface. - const exportMatch = distDecl.match(/export\s*\{([^}]*)\}/); - expect(exportMatch).not.toBeNull(); - const exportedTypeNames = exportMatch![1] - .split(',') - .map((entry) => entry.trim()) - .filter((entry) => entry.startsWith('type ')) - .map((entry) => entry.slice('type '.length).trim()) - .filter(Boolean); - const expectedTypes = [...TYPES_EXPORTS, ...REPORTER_EXPORTS, ...METRICS_EXPORTS]; + const expectedTypeSet = new Set(expectedTypes); + + const exportedTypeNames: string[] = []; + + // export { type A, B } or export { A, B } from 'pkg' + for (const match of distDecl.matchAll(/export\s+(?:type\s*)?\{([^}]+)\}(?:\s*from\s*['"][^'"]+['"])?\s*;?/g)) { + const blockIsTypeOnly = match[0].startsWith('export type {'); + for (let entry of match[1].split(',')) { + entry = entry.trim(); + if (!entry) continue; + // strip optional `type` prefix and `as` aliases + const name = entry + .replace(/^type\s+/, '') + .replace(/\s+as\s+.*/, '') + .trim(); + if (entry.startsWith('type ') || blockIsTypeOnly) { + exportedTypeNames.push(name); + } else if (expectedTypeSet.has(name)) { + // Re-exports from workspace packages are not always prefixed with `type` + // in declaration emit. Treat names that are part of the expected + // public type surface as type exports. + exportedTypeNames.push(name); + } + } + } + + // export interface X / export type X = ... (top-level declarations) + for (const match of distDecl.matchAll(/export\s+(?:type\s+(\w+)|interface\s+(\w+))/g)) { + exportedTypeNames.push(match[1] ?? match[2]); + } + + const uniqueExportedTypeNames = [...new Set(exportedTypeNames)].sort(); + // The exported type set must match the expected surface exactly. - expect(exportedTypeNames.slice().sort()).toEqual([...expectedTypes].sort()); + expect(uniqueExportedTypeNames).toEqual([...expectedTypes].sort()); // Every expected type must also appear as a declaration in the .d.ts. for (const name of expectedTypes) { @@ -1448,7 +1470,7 @@ describe('public API surface and type exports', () => { // Internal helper types must never be surfaced by the built barrel. for (const internal of INTERNAL_TYPES) { - expect(exportedTypeNames).not.toContain(internal); + expect(uniqueExportedTypeNames).not.toContain(internal); } // The internal types do exist in types.ts, they are just not surfaced. diff --git a/packages/benchsdk/src/client.ts b/packages/benchsdk/src/client.ts index e4cb9c76..bac3f54a 100644 --- a/packages/benchsdk/src/client.ts +++ b/packages/benchsdk/src/client.ts @@ -1,827 +1,15 @@ -import type { - BenchmarkAssignment, - BenchmarkArtifact, - BenchmarkClient, - BenchmarkClientConfig, - BenchmarkParticipant, - BenchmarkResource, - BenchmarkRun, - BenchmarkResultsOverview, - BenchmarkResultsOverviewInput, - BenchmarkRunImports, - BenchmarkRunResults, - BenchmarkRunSummaryInput, - BenchmarkRunTaskResults, - BenchmarkRunTaskResultsInput, - BenchmarkRunTimeline, - BenchmarkRunTimelineInput, - BenchmarkRunWorker, - BenchmarkWorkerAttempt, - CreateWorkerArtifactInput, - CreateWorkerArtifactResponse, - ClaimWorkerInput, - CreateRunInput, - DefineStepOptions, - JsonObject, - RunWorkerOptions, - RunWorkerResult, - SendTaskResultsInput, - PlanWorkersInput, - TaskStepRecord, - TaskResultRecord, - TaskResultsResponse, - RunProgress, - UpdateBenchmarkInput, - UpdateParticipantInput, - UpdateRunInput, - UpdateWorkerInput, - UpsertBenchmarkInput, - UpsertParticipantInput, - UploadWorkerArtifactInput, - WorkerConcurrencySample, - WorkerHeartbeatInput, -} from './types'; +import { createBenchmarkClient as createApiClient, BenchmarkApiError } from '@benchsdk/api'; +import { runWorker } from '@benchsdk/worker'; +import type { BenchmarkClient, BenchmarkClientConfig, RunWorkerOptions, RunWorkerResult } from './types'; -const DEFAULT_BASE_URL = 'https://platform.computesdk.com/api/v1'; -const DEFAULT_BATCH_SIZE = 1000; -const DEFAULT_HEARTBEAT_INTERVAL_MS = 30_000; -const DEFAULT_FLUSH_INTERVAL_MS = 30_000; -const DEFAULT_READY_POLL_INTERVAL_MS = 1000; -const MAX_TASK_RESULT_RECORDS = 5000; -const MAX_TASK_RECORD_STEPS = 100; -const MAX_HEARTBEAT_CONCURRENCY_SAMPLES = 20; -const MAX_WORKER_LOG_LINES = 100_000; - -export class BenchmarkApiError extends Error { - constructor( - message: string, - public readonly status: number, - public readonly body: string, - ) { - super(message); - this.name = 'BenchmarkApiError'; - } -} - -function trimTrailingSlash(value: string): string { - return value.replace(/\/+$/, ''); -} - -function encodePath(value: string): string { - return encodeURIComponent(value); -} - -function queryString(input: Record): string { - const params = new URLSearchParams(); - for (const [key, value] of Object.entries(input)) { - if (value !== undefined) params.set(key, String(value)); - } - const value = params.toString(); - return value ? `?${value}` : ''; -} - -function getApiKey(input?: string): string | undefined { - return input ?? (typeof process !== 'undefined' ? process.env.COMPUTESDK_ADMIN_API_KEY ?? process.env.COMPUTESDK_API_KEY : undefined); -} - -function getErrorCode(error: unknown): string { - if (error instanceof Error && 'code' in error && typeof (error as { code: unknown }).code === 'string' && (error as { code: string }).code) { - return (error as { code: string }).code; - } - if (error instanceof Error && error.name) return error.name; - return 'ERROR'; -} - -function toJsonObject(value: unknown): JsonObject | undefined { - if (!value || typeof value !== 'object' || Array.isArray(value)) return undefined; - return value as JsonObject; -} - -/** Merges task-level measurements with a returned data payload; undefined when empty. */ -function mergeMeasures(measures: JsonObject, returned: JsonObject | undefined): JsonObject | undefined { - const merged = { ...measures, ...(returned ?? {}) }; - return Object.keys(merged).length > 0 ? merged : undefined; -} - -/** Synthesizes the single step that represents a task that declared no steps. */ -function implicitTaskStep(record: TaskResultRecord, measures: JsonObject): TaskStepRecord { - return { - name: 'task', - status: record.status === 'success' ? 'success' : 'error', - startedAt: record.startedAt, - completedAt: record.completedAt, - latencyMs: record.latencyMs, - errorCode: record.errorCode ?? null, - data: Object.keys(measures).length > 0 ? { ...measures } : undefined, - }; -} - -function validateTaskResults(input: SendTaskResultsInput): void { - if (input.records.length > MAX_TASK_RESULT_RECORDS) { - throw new Error(`Benchmark task result batches are limited to ${MAX_TASK_RESULT_RECORDS} records.`); - } - - for (const record of input.records) { - if ((record.steps?.length ?? 0) > MAX_TASK_RECORD_STEPS) { - throw new Error(`Benchmark task result records are limited to ${MAX_TASK_RECORD_STEPS} steps.`); - } - } -} - -function validateHeartbeat(input: WorkerHeartbeatInput): void { - const concurrency = input.concurrency ?? []; - if (concurrency.length > MAX_HEARTBEAT_CONCURRENCY_SAMPLES) { - throw new Error(`Benchmark heartbeat concurrency is limited to ${MAX_HEARTBEAT_CONCURRENCY_SAMPLES} samples.`); - } - - const steps = new Set(); - for (const sample of concurrency) { - if (steps.has(sample.step)) { - throw new Error(`Benchmark heartbeat concurrency step values must be unique per heartbeat.`); - } - steps.add(sample.step); - } -} - -function validatePositiveInteger(name: string, value: number): void { - if (!Number.isInteger(value) || value <= 0) { - throw new Error(`Benchmark ${name} must be a positive integer.`); - } -} - -function validateBatchSize(value: number): void { - validatePositiveInteger('batchSize', value); - if (value > MAX_TASK_RESULT_RECORDS) { - throw new Error(`Benchmark batchSize must be at most ${MAX_TASK_RESULT_RECORDS}.`); - } -} - -function normalizeArtifacts(data: { items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }): BenchmarkArtifact[] { - return data.items ?? data.artifacts ?? []; -} - -function bodySizeBytes(body: UploadWorkerArtifactInput['body']): number | undefined { - if (typeof body === 'string') return new TextEncoder().encode(body).byteLength; - if (body instanceof Blob) return body.size; - if (body instanceof ArrayBuffer) return body.byteLength; - if (ArrayBuffer.isView(body)) return body.byteLength; - if (body instanceof URLSearchParams) return new TextEncoder().encode(body.toString()).byteLength; - return undefined; -} - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -async function mapPool(items: T[], concurrency: number, fn: (item: T) => Promise): Promise { - let nextIndex = 0; - const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => { - while (nextIndex < items.length) { - const item = items[nextIndex]; - nextIndex += 1; - await fn(item); - } - }); - await Promise.all(workers); -} +export { BenchmarkApiError }; export function createBenchmarkClient(config: BenchmarkClientConfig = {}): BenchmarkClient { - const baseUrl = trimTrailingSlash(config.baseUrl ?? DEFAULT_BASE_URL); - const apiKey = getApiKey(config.apiKey); - const fetchImpl = config.fetch ?? (typeof fetch !== 'undefined' ? fetch : undefined); - - if (!fetchImpl) { - throw new Error('fetch is not available'); - } - const doFetch = fetchImpl; - - async function request(method: string, path: string, body?: JsonObject): Promise { - const headers: Record = { 'Content-Type': 'application/json' }; - if (apiKey) headers.Authorization = `Bearer ${apiKey}`; - - const response = await doFetch(`${baseUrl}${path}`, { - method, - headers, - body: body === undefined ? undefined : JSON.stringify(body), - }); - const text = await response.text(); - - if (!response.ok) { - throw new BenchmarkApiError( - `Benchmark API request failed: ${response.status} ${response.statusText}`, - response.status, - text, - ); - } - - return (text ? JSON.parse(text) : {}) as T; - } - - async function sendTaskResults(input: SendTaskResultsInput): Promise { - if (input.records.length === 0) { - return {}; - } - validateTaskResults(input); - - return request( - 'POST', - `/benchmarks/${encodePath(input.benchmarkSlug)}/runs/${encodePath(input.runId)}/workers/${encodePath(input.workerId)}/events`, - { - type: 'task_results', - attemptId: input.attemptId, - sequenceNumber: input.sequenceNumber, - isFinal: input.isFinal, - records: input.records as unknown as JsonObject[], - }, - ); - } - - async function updateWorkerLifecycle( - action: 'heartbeat' | 'complete' | 'fail' | 'release', - benchmarkSlug: string, - runId: string, - workerId: string, - attemptId: string, - extra?: JsonObject, - ): Promise<{ worker: BenchmarkRunWorker; attempt: BenchmarkWorkerAttempt }> { - return request<{ worker: BenchmarkRunWorker; attempt: BenchmarkWorkerAttempt }>( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/${action}`, - { attemptId, ...(extra ?? {}) }, - ); - } - - const client: BenchmarkClient = { - async upsertBenchmark(slug, input) { - const data = await request<{ benchmark: BenchmarkResource }>('PUT', `/benchmarks/${encodePath(slug)}`, input as unknown as JsonObject); - return data.benchmark; - }, - - async getBenchmark(slug) { - const data = await request<{ benchmark: BenchmarkResource }>('GET', `/benchmarks/${encodePath(slug)}`); - return data.benchmark; - }, - - async updateBenchmark(slug, input: UpdateBenchmarkInput) { - const data = await request<{ benchmark: BenchmarkResource }>('PATCH', `/benchmarks/${encodePath(slug)}`, input as unknown as JsonObject); - return data.benchmark; - }, - - async listBenchmarks() { - const data = await request<{ items?: BenchmarkResource[]; benchmarks?: BenchmarkResource[] }>('GET', '/benchmarks'); - return data.items ?? data.benchmarks ?? []; - }, - - async createRun(benchmarkSlug, input) { - return request<{ run: BenchmarkRun; participants: BenchmarkParticipant[]; organizationSlug: string }>( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs`, - input as unknown as JsonObject, - ); - }, - - async listRuns(benchmarkSlug) { - const data = await request<{ items: BenchmarkRun[] }>('GET', `/benchmarks/${encodePath(benchmarkSlug)}/runs`); - return data.items; - }, - - async getRun(benchmarkSlug, runId) { - const data = await request<{ run: BenchmarkRun }>('GET', `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}`); - return data.run; - }, - - async updateRun(benchmarkSlug, runId, input: UpdateRunInput) { - const data = await request<{ run: BenchmarkRun }>( - 'PATCH', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}`, - input as unknown as JsonObject, - ); - return data.run; - }, - - async upsertParticipant(benchmarkSlug, runId, participantSlug, input = {}) { - const data = await request<{ participant: BenchmarkParticipant }>( - 'PUT', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, - input as JsonObject, - ); - return data.participant; - }, - - async listParticipants(benchmarkSlug, runId) { - const data = await request<{ items?: BenchmarkParticipant[]; participants?: BenchmarkParticipant[] }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants`, - ); - return data.items ?? data.participants ?? []; - }, - - async getParticipant(benchmarkSlug, runId, participantSlug) { - const data = await request<{ participant: BenchmarkParticipant }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, - ); - return data.participant; - }, - - async updateParticipant(benchmarkSlug, runId, participantSlug, input: UpdateParticipantInput) { - const data = await request<{ participant: BenchmarkParticipant }>( - 'PATCH', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}`, - input as unknown as JsonObject, - ); - return data.participant; - }, - - async getRunProgress(benchmarkSlug, runId) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/progress`, - ); - }, - - async listWorkers(benchmarkSlug, runId, participantSlug) { - const data = await request<{ items?: BenchmarkRunWorker[]; workers?: BenchmarkRunWorker[] }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers`, - ); - return data.items ?? data.workers ?? []; - }, - - async planWorkers(benchmarkSlug, runId, participantSlug, input: PlanWorkersInput = {}) { - const data = await request<{ items?: BenchmarkRunWorker[]; workers?: BenchmarkRunWorker[] }>( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers`, - input as JsonObject, - ); - return data.items ?? data.workers ?? []; - }, - - async getWorker(benchmarkSlug, runId, workerId) { - const data = await request<{ worker: BenchmarkRunWorker }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}`, - ); - return data.worker; - }, - - async updateWorker(benchmarkSlug, runId, workerId, input: UpdateWorkerInput) { - const data = await request<{ worker: BenchmarkRunWorker }>( - 'PATCH', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}`, - input as unknown as JsonObject, - ); - return data.worker; - }, - - async claimWorker(benchmarkSlug, runId, participantSlug, input: ClaimWorkerInput = {}) { - const data = await request<{ assignment: BenchmarkAssignment | null }>( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/participants/${encodePath(participantSlug)}/workers/claim`, - input as JsonObject, - ); - return data.assignment; - }, - - sendTaskResults, - - async heartbeatWorker(benchmarkSlug, runId, workerId, input: WorkerHeartbeatInput) { - validateHeartbeat(input); - const { attemptId, ...extra } = input; - if (extra.currentStep == null) { - delete extra.currentStep; - } - return updateWorkerLifecycle('heartbeat', benchmarkSlug, runId, workerId, attemptId, extra as JsonObject); - }, - - releaseWorker(benchmarkSlug, runId, workerId, attemptId) { - return updateWorkerLifecycle('release', benchmarkSlug, runId, workerId, attemptId); - }, - - completeWorker(benchmarkSlug, runId, workerId, attemptId) { - return updateWorkerLifecycle('complete', benchmarkSlug, runId, workerId, attemptId); - }, - - failWorker(benchmarkSlug, runId, workerId, attemptId, error) { - return updateWorkerLifecycle('fail', benchmarkSlug, runId, workerId, attemptId, { - errorCode: getErrorCode(error), - errorMessage: error instanceof Error ? error.message : String(error ?? 'Unknown error'), - }); - }, - - async createWorkerArtifact(benchmarkSlug, runId, workerId, input: CreateWorkerArtifactInput) { - return request( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/artifacts`, - input as unknown as JsonObject, - ); - }, - - async uploadWorkerArtifact(benchmarkSlug, runId, workerId, input: UploadWorkerArtifactInput) { - const sizeBytes = bodySizeBytes(input.body); - const artifactInput: CreateWorkerArtifactInput = { - attemptId: input.attemptId, - kind: input.kind, - contentType: input.contentType, - name: input.name, - metadata: sizeBytes === undefined ? input.metadata : { ...input.metadata, sizeBytes }, - }; - const response = await client.createWorkerArtifact(benchmarkSlug, runId, workerId, artifactInput); - const uploadUrl = response.uploadUrl ?? response.artifact?.uploadUrl; - if (!uploadUrl) { - throw new Error('Benchmark artifact upload URL is missing.'); - } - const uploadResponse = await doFetch(uploadUrl, { - method: 'PUT', - headers: input.contentType ? { 'Content-Type': input.contentType } : undefined, - body: input.body, - }); - if (!uploadResponse.ok) { - const errorBody = await uploadResponse.text().catch(() => ''); - throw new BenchmarkApiError( - `Benchmark artifact upload failed with ${uploadResponse.status}`, - uploadResponse.status, - errorBody, - ); - } - return response; - }, - - async listRunArtifacts(benchmarkSlug, runId) { - const data = await request<{ items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/artifacts`, - ); - return normalizeArtifacts(data); - }, - - async listWorkerArtifacts(benchmarkSlug, runId, workerId) { - const data = await request<{ items?: BenchmarkArtifact[]; artifacts?: BenchmarkArtifact[] }>( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/workers/${encodePath(workerId)}/artifacts`, - ); - return normalizeArtifacts(data); - }, - - async getBenchmarkResults(benchmarkSlug, input: BenchmarkResultsOverviewInput = {}) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/results${queryString({ limit: input.limit })}`, - ); - }, - - async getRunResults(benchmarkSlug, runId) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results`, - ); - }, - - async getRunTaskResults(benchmarkSlug, runId, input: BenchmarkRunTaskResultsInput = {}) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/tasks${queryString({ bucketSize: input.bucketSize, failureLimit: input.failureLimit })}`, - ); - }, - - async getRunTimeline(benchmarkSlug, runId, input: BenchmarkRunTimelineInput = {}) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/timeline${queryString({ bucketMs: input.bucketMs })}`, - ); - }, - - async getRunImports(benchmarkSlug, runId) { - return request( - 'GET', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/results/imports`, - ); - }, - - async submitRunSummary(benchmarkSlug, runId, input) { - await request( - 'POST', - `/benchmarks/${encodePath(benchmarkSlug)}/runs/${encodePath(runId)}/summary`, - input as unknown as JsonObject, - ); - }, - - async runWorker(options: RunWorkerOptions): Promise { - if (options.concurrency !== undefined) validatePositiveInteger('concurrency', options.concurrency); - if (options.batchSize !== undefined) validateBatchSize(options.batchSize); - if (options.flushIntervalMs !== undefined) validatePositiveInteger('flushIntervalMs', options.flushIntervalMs); - - const assignment = await client.claimWorker(options.benchmarkSlug, options.runId, options.participantSlug, { - processKind: options.processKind, - processKey: options.processKey, - }); - if (!assignment) return { assignment: null, records: [] }; - const claimed = assignment; - - let sequenceNumber = 0; - const records: TaskResultRecord[] = []; - const pending: TaskResultRecord[] = []; - const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE; - const workerConcurrency = options.concurrency ?? claimed.targetConcurrency; - validatePositiveInteger('concurrency', workerConcurrency); - const taskIndices = Array.from({ length: claimed.taskRange.count }, (_, index) => claimed.taskRange.start + index); - const activeByStep = new Map(); - const targetByStep = new Map(); - const readyWaitByStep = new Map>(); - let doneCount = 0; - let errorCount = 0; - let inFlightCount = 0; - let flushChain = Promise.resolve(); - - function concurrencySamples(): WorkerConcurrencySample[] { - return Array.from(activeByStep.entries()) - .filter(([, active]) => active > 0) - .map(([step, active]) => ({ - step, - active, - target: targetByStep.get(step) ?? workerConcurrency, - })) - .sort((a, b) => b.active - a.active) - .slice(0, MAX_HEARTBEAT_CONCURRENCY_SAMPLES); - } - - async function sendHeartbeat(): Promise { - const concurrency = concurrencySamples(); - const step = concurrency[0]?.step ?? null; - await client.heartbeatWorker(options.benchmarkSlug, options.runId, claimed.workerId, { - attemptId: claimed.attemptId, - progressDone: doneCount, - progressInFlight: inFlightCount, - progressErrors: errorCount, - progressTotal: taskIndices.length, - ...(step ? { currentStep: step } : {}), - concurrency, - }); - } - - let heartbeatInFlight: Promise | null = null; - let heartbeatRequested = false; - - function requestHeartbeat(): void { - heartbeatRequested = true; - if (heartbeatInFlight) return; - - heartbeatInFlight = (async () => { - while (heartbeatRequested) { - heartbeatRequested = false; - await sendHeartbeat().catch(() => {}); - } - })().finally(() => { - heartbeatInFlight = null; - if (heartbeatRequested) requestHeartbeat(); - }); - } - - async function pollStepReady(stepName: string, stepOptions: DefineStepOptions): Promise { - const startedAt = Date.now(); - const pollInterval = stepOptions.readyPollIntervalMs ?? options.readyPollIntervalMs ?? DEFAULT_READY_POLL_INTERVAL_MS; - - while (true) { - const progress = await client.getRunProgress(options.benchmarkSlug, options.runId); - const participant = progress.participants.find((item) => item.slug === options.participantSlug); - const step = participant?.concurrency.find((item) => item.step === stepName); - if (step?.ready) return; - - if (typeof stepOptions.readyTimeoutMs === 'number' && Date.now() - startedAt >= stepOptions.readyTimeoutMs) { - throw new Error(`Timed out waiting for benchmark step "${stepName}" to become ready.`); - } - - await sleep(pollInterval); - } - } - - async function waitForStepReady(stepName: string, stepOptions: DefineStepOptions): Promise { - const existing = readyWaitByStep.get(stepName); - if (existing) return existing; - - const wait = pollStepReady(stepName, stepOptions).finally(() => { - if (readyWaitByStep.get(stepName) === wait) readyWaitByStep.delete(stepName); - }); - readyWaitByStep.set(stepName, wait); - return wait; - } - - const heartbeat = setInterval(() => { - requestHeartbeat(); - }, options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS); - heartbeat.unref?.(); - - async function flush(isFinal: boolean, force = false): Promise { - flushChain = flushChain.then(async () => { - if (force && doneCount >= taskIndices.length) return; - while (pending.length >= batchSize || ((isFinal || force) && pending.length > 0)) { - const batch = pending.splice(0, batchSize); - await sendTaskResults({ - benchmarkSlug: options.benchmarkSlug, - runId: options.runId, - workerId: claimed.workerId, - attemptId: claimed.attemptId, - sequenceNumber, - isFinal: isFinal && pending.length === 0, - records: batch, - }); - sequenceNumber += 1; - } - }); - await flushChain; - } - - const resultFlush = setInterval(() => { - if (doneCount < taskIndices.length) void flush(false, true).catch(() => {}); - }, options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS); - resultFlush.unref?.(); - - // Accumulated across the worker's tasks via `ctx.step` and `ctx.log`, - // uploaded once as a worker log artifact when the worker finishes. - const workerLogLines: string[] = []; - let workerLogTruncated = false; - function appendWorkerLog(line: string): void { - if (workerLogLines.length >= MAX_WORKER_LOG_LINES) { - if (!workerLogTruncated) { - workerLogTruncated = true; - workerLogLines.push('... (worker log truncated)'); - } - return; - } - workerLogLines.push(line); - } - - async function runFinishHook(status: 'success' | 'error'): Promise { - await options.onFinish?.({ - assignment: claimed, - records, - status, - client, - uploadArtifact(input) { - return client.uploadWorkerArtifact(options.benchmarkSlug, options.runId, claimed.workerId, { - ...input, - attemptId: claimed.attemptId, - }); - }, - }); - } - - async function uploadWorkerLogArtifact(): Promise { - if (workerLogLines.length === 0) return; - try { - await client.uploadWorkerArtifact(options.benchmarkSlug, options.runId, claimed.workerId, { - attemptId: claimed.attemptId, - kind: 'coordinator.log', - contentType: 'text/plain', - name: 'worker.log', - body: workerLogLines.join('\n') + '\n', - }); - } catch { - // Log upload is best-effort; never fail the run over it. - } - } - - try { - await sendHeartbeat().catch(() => {}); - - await mapPool(taskIndices, workerConcurrency, async (taskIndex) => { - inFlightCount += 1; - const startedAtDate = new Date(); - const startedAtMs = Date.now(); - const record: TaskResultRecord = { - taskIndex, - status: 'success', - startedAt: startedAtDate.toISOString(), - }; - const steps: TaskStepRecord[] = []; - const taskMeasures: JsonObject = {}; - // The step a `measure(...)` call currently attributes to. Set while a - // step's fn runs; null at task top-level (measures go on the record). - let activeStep: TaskStepRecord | null = null; - - function measure(data: JsonObject): void { - if (activeStep) { - activeStep.data = { ...(activeStep.data ?? {}), ...data }; - } else { - Object.assign(taskMeasures, data); - } - } - - function log(message: string, meta?: JsonObject): void { - const suffix = meta && Object.keys(meta).length > 0 ? ` ${JSON.stringify(meta)}` : ''; - appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${message}${suffix}`); - } - - async function step(name: string, fn: () => Promise | T, stepOptions: DefineStepOptions = {}): Promise { - const stepStartedAtMs = Date.now(); - const stepRecord: TaskStepRecord = { - name, - status: 'success', - startedAt: new Date().toISOString(), - completedAt: new Date().toISOString(), - latencyMs: 0, - }; - if (stepOptions.timeoutMs !== undefined) stepRecord.timeoutMs = stepOptions.timeoutMs; - if (stepOptions.stepConcurrency !== undefined) stepRecord.concurrency = stepOptions.stepConcurrency; - - const shouldReportConcurrency = stepOptions.reportConcurrency ?? true; - if (shouldReportConcurrency) { - const stepConcurrency = stepOptions.concurrency ?? workerConcurrency; - validatePositiveInteger(`step "${name}" concurrency`, stepConcurrency); - targetByStep.set(name, stepConcurrency); - activeByStep.set(name, (activeByStep.get(name) ?? 0) + 1); - requestHeartbeat(); - } - - const previousStep = activeStep; - activeStep = stepRecord; - try { - if (stepOptions.readiness === 'poll') { - await waitForStepReady(name, stepOptions); - } - const value = await fn(); - appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${name}`); - return value; - } catch (error) { - stepRecord.status = 'error'; - stepRecord.errorCode = getErrorCode(error); - appendWorkerLog(`${new Date().toISOString()} [task ${taskIndex}] ${name}`); - appendWorkerLog(` error: ${error instanceof Error ? error.message : String(error)}`); - throw error; - } finally { - activeStep = previousStep; - stepRecord.completedAt = new Date().toISOString(); - stepRecord.latencyMs = Date.now() - stepStartedAtMs; - steps.push(stepRecord); - if (shouldReportConcurrency) { - const nextActive = Math.max(0, (activeByStep.get(name) ?? 0) - 1); - if (nextActive === 0) { - activeByStep.delete(name); - targetByStep.delete(name); - } else { - activeByStep.set(name, nextActive); - } - requestHeartbeat(); - } - } - } - - try { - const data = await options.task({ assignment: claimed, taskIndex, step, measure, log }); - record.data = mergeMeasures(taskMeasures, toJsonObject(data)); - } catch (error) { - record.status = 'error'; - record.errorCode = getErrorCode(error); - record.data = mergeMeasures(taskMeasures, { errorMessage: error instanceof Error ? error.message : String(error) }); - } finally { - record.completedAt = new Date().toISOString(); - record.latencyMs = Date.now() - startedAtMs; - // A task with no explicit steps is recorded as a single implicit - // 'task' step, so every task contributes at least one step row. - if (steps.length === 0) { - steps.push(implicitTaskStep(record, taskMeasures)); - } - record.steps = steps; - doneCount += 1; - inFlightCount = Math.max(0, inFlightCount - 1); - if (record.status !== 'success') errorCount += 1; - } - - records.push(record); - pending.push(record); - options.onResult?.(record); - if (pending.length >= batchSize) await flush(false); - }); - - await flush(true); - - const hasErrors = records.some((record) => record.status !== 'success'); - try { - await runFinishHook(hasErrors ? 'error' : 'success'); - } catch (error) { - if (!hasErrors) throw error; - } - - if (hasErrors) { - await client.failWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId, new Error('One or more tasks failed')); - } else { - await client.completeWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId); - } - - return { assignment: claimed, records }; - } catch (error) { - await flush(true).catch(() => {}); - await runFinishHook('error').catch(() => {}); - await client.failWorker(options.benchmarkSlug, options.runId, claimed.workerId, claimed.attemptId, error).catch(() => {}); - throw error; - } finally { - await uploadWorkerLogArtifact(); - clearInterval(heartbeat); - clearInterval(resultFlush); - } + const apiClient = createApiClient(config); + return { + ...apiClient, + runWorker(options: RunWorkerOptions): Promise { + return runWorker(apiClient, options as unknown as any) as Promise; }, - }; - - return client; + } as BenchmarkClient; } - diff --git a/packages/benchsdk/src/metrics.ts b/packages/benchsdk/src/metrics.ts index 77dcec14..da2f84c5 100644 --- a/packages/benchsdk/src/metrics.ts +++ b/packages/benchsdk/src/metrics.ts @@ -1,93 +1,5 @@ -import * as fs from 'node:fs'; -import * as os from 'node:os'; -import { monitorEventLoopDelay } from 'node:perf_hooks'; - -export interface BenchmarkSystemMetricsSample { - ts: string; - uptimeMs: number; - cpuUserUs: number; - cpuSystemUs: number; - memRssMb: number; - memHeapUsedMb: number; - memHeapTotalMb: number; - memExternalMb: number; - eventLoopP50Ms: number; - eventLoopP99Ms: number; - eventLoopMaxMs: number; - loadavg1m: number; - loadavg5m: number; - loadavg15m: number; - openFds: number | null; - sockstat: Record | null; -} - -export interface BenchmarkSystemMetricsCollector { - sample(): BenchmarkSystemMetricsSample; - stop(): void; -} - -function readSockstat(): Record | null { - try { - const data = fs.readFileSync('/proc/net/sockstat', 'utf-8'); - const out: Record = {}; - for (const line of data.split('\n')) { - const index = line.indexOf(':'); - if (index < 0) continue; - const section = line.slice(0, index).trim().toLowerCase(); - const parts = line.slice(index + 1).trim().split(/\s+/); - for (let i = 0; i + 1 < parts.length; i += 2) { - const value = Number.parseInt(parts[i + 1], 10); - if (!Number.isNaN(value)) out[`${section}_${parts[i]}`] = value; - } - } - return out; - } catch { - return null; - } -} - -function countOpenFds(): number | null { - try { - return fs.readdirSync('/proc/self/fd').length; - } catch { - return null; - } -} - -export function createSystemMetricsCollector(): BenchmarkSystemMetricsCollector { - const startedAt = Date.now(); - const cpuBaseline = process.cpuUsage(); - const eventLoop = monitorEventLoopDelay({ resolution: 20 }); - eventLoop.enable(); - - return { - sample() { - const cpu = process.cpuUsage(cpuBaseline); - const memory = process.memoryUsage(); - const loadavg = os.loadavg(); - const sample: BenchmarkSystemMetricsSample = { - ts: new Date().toISOString(), - uptimeMs: Date.now() - startedAt, - cpuUserUs: cpu.user, - cpuSystemUs: cpu.system, - memRssMb: Math.round(memory.rss / 1024 / 1024), - memHeapUsedMb: Math.round(memory.heapUsed / 1024 / 1024), - memHeapTotalMb: Math.round(memory.heapTotal / 1024 / 1024), - memExternalMb: Math.round(memory.external / 1024 / 1024), - eventLoopP50Ms: eventLoop.percentile(50) / 1e6, - eventLoopP99Ms: eventLoop.percentile(99) / 1e6, - eventLoopMaxMs: eventLoop.max / 1e6, - loadavg1m: loadavg[0], - loadavg5m: loadavg[1], - loadavg15m: loadavg[2], - openFds: countOpenFds(), - sockstat: readSockstat(), - }; - eventLoop.reset(); - return sample; - }, - stop() { - eventLoop.disable(); - }, - }; -} +export { createSystemMetricsCollector } from '@benchsdk/worker'; +export type { + BenchmarkSystemMetricsCollector, + BenchmarkSystemMetricsSample, +} from '@benchsdk/worker'; diff --git a/packages/benchsdk/src/participants.ts b/packages/benchsdk/src/participants.ts index 806acfa8..54fadd41 100644 --- a/packages/benchsdk/src/participants.ts +++ b/packages/benchsdk/src/participants.ts @@ -1,54 +1,2 @@ -/** - * Base interface for a benchmark participant — the shared shape across all - * benchmark categories (sandbox, ai-gateway, browser, storage). Each category - * extends this with its own provider-specific fields. - */ -export interface BaseParticipant { - /** Participant name (e.g. 'e2b', 'daytona', 'openrouter') */ - name: string; - /** Environment variables that must all be set to run this participant */ - requiredEnvVars: string[]; -} - -/** -/** - * Filters `participants` down to those whose `requiredEnvVars` are all set - * in `process.env`. Returns an object `{ available, skipped }` where `skipped` - * includes the names and missing vars for logging. - */ -export function filterParticipantsByEnv( - participants: T[], -): { available: T[]; skipped: { name: string; missing: string[] }[] } { - const available: T[] = []; - const skipped: { name: string; missing: string[] }[] = []; - - for (const p of participants) { - const missing = p.requiredEnvVars.filter((v) => !process.env[v]); - if (missing.length > 0) { - skipped.push({ name: p.name, missing }); - } else { - available.push(p); - } - } - - return { available, skipped }; -} - -/** - * Filters `all` down to the requested `names`, exiting with a clear error - * if any name is unrecognized. Returns `all` unchanged when `names` is - * undefined (no filter specified). - */ -export function selectParticipants( - all: T[], - names?: string[], -): T[] { - if (!names) return all; - const unknown = names.filter((n) => !all.some((p) => p.name === n)); - if (unknown.length > 0) { - console.error(`Unknown participant(s): ${unknown.join(', ')}`); - console.error(`Available: ${all.map((p) => p.name).join(', ')}`); - process.exit(1); - } - return all.filter((p) => names.includes(p.name)); -} +export { filterParticipantsByEnv, selectParticipants } from '@benchsdk/worker'; +export type { BaseParticipant } from '@benchsdk/worker'; diff --git a/packages/benchsdk/src/reporter.ts b/packages/benchsdk/src/reporter.ts index 147e4c89..f1f5c68f 100644 --- a/packages/benchsdk/src/reporter.ts +++ b/packages/benchsdk/src/reporter.ts @@ -1,238 +1,9 @@ -import { createBenchmarkClient } from './client'; -import type { - BenchmarkAssignment, - BenchmarkClient, - BenchmarkClientConfig, - CreateWorkerArtifactResponse, - JsonObject, - TaskResultRecord, - WorkerConcurrencySample, -} from './types'; - -const DEFAULT_REPORTER_BATCH_SIZE = 500; -const DEFAULT_READY_POLL_INTERVAL_MS = 1000; - -export interface BenchmarkReporterConfig extends BenchmarkClientConfig { - benchmarkSlug: string; - runId: string; - participantSlug: string; - processKind?: string; - processKey?: string; - batchSize?: number; -} - -export interface BenchmarkReporterProgress { - done: number; - inFlight: number; - errors: number; - total?: number; -} - -export interface BenchmarkReporterArtifactInput { - kind: string; - name?: string; - contentType?: string; - body: BodyInit; - metadata?: JsonObject; -} - -export interface BenchmarkReporterHeartbeatInput { - currentStep?: string | null; - concurrency?: WorkerConcurrencySample[]; -} - -export interface BenchmarkReporterBarrierInput { - step: string; - timeoutMs?: number; - pollIntervalMs?: number; - active?: number; - target?: number; - concurrency?: WorkerConcurrencySample[]; -} - -export interface BenchmarkReporterBarrierResult { - active: number | null; - target: number | null; - ready: boolean; - measuredAt: string; -} - -export class BenchmarkReporter { - private readonly client: BenchmarkClient; - private readonly assignment: BenchmarkAssignment; - private readonly cfg: Required>; - private pending: TaskResultRecord[] = []; - private sequenceNumber = 0; - private flushChain: Promise = Promise.resolve(); - private progress: BenchmarkReporterProgress; - private barrier: { step: string; concurrency: WorkerConcurrencySample[] } | null = null; - - private constructor(client: BenchmarkClient, cfg: BenchmarkReporterConfig, assignment: BenchmarkAssignment) { - this.client = client; - this.assignment = assignment; - this.cfg = { - benchmarkSlug: cfg.benchmarkSlug, - runId: cfg.runId, - participantSlug: cfg.participantSlug, - batchSize: cfg.batchSize ?? DEFAULT_REPORTER_BATCH_SIZE, - }; - this.progress = { done: 0, inFlight: 0, errors: 0, total: assignment.taskRange.count }; - } - - static async claim(cfg: BenchmarkReporterConfig): Promise { - const client = createBenchmarkClient(cfg); - try { - const assignment = await client.claimWorker(cfg.benchmarkSlug, cfg.runId, cfg.participantSlug, { - processKind: cfg.processKind, - processKey: cfg.processKey, - }); - return assignment ? new BenchmarkReporter(client, cfg, assignment) : null; - } catch (error) { - console.warn( - `[benchsdk] failed to claim worker for ${cfg.benchmarkSlug}/${cfg.participantSlug}: ` + - `${error instanceof Error ? error.message : String(error)}`, - ); - return null; - } - } - - get workerAssignment(): BenchmarkAssignment { - return this.assignment; - } - - get taskCount(): number { - return this.assignment.taskRange.count; - } - - get taskIndexStart(): number { - return this.assignment.taskRange.start; - } - - setProgress(progress: BenchmarkReporterProgress): void { - this.progress = { ...progress, total: progress.total ?? this.assignment.taskRange.count }; - } - - recordResult(record: TaskResultRecord): void { - this.pending.push(record); - if (this.pending.length >= this.cfg.batchSize) void this.flush(false); - } - - async heartbeat(input: BenchmarkReporterHeartbeatInput = {}): Promise { - const barrier = this.barrier; - const currentStep = barrier?.step ?? input.currentStep; - const concurrency = barrier?.concurrency ?? input.concurrency; - await this.client.heartbeatWorker(this.cfg.benchmarkSlug, this.cfg.runId, this.assignment.workerId, { - attemptId: this.assignment.attemptId, - progressDone: this.progress.done, - progressInFlight: this.progress.inFlight, - progressErrors: this.progress.errors, - progressTotal: this.progress.total, - ...(currentStep ? { currentStep } : {}), - ...(concurrency ? { concurrency } : {}), - }).catch(() => {}); - } - - async waitForStepReady(input: BenchmarkReporterBarrierInput): Promise { - const startedAt = Date.now(); - const pollIntervalMs = input.pollIntervalMs ?? DEFAULT_READY_POLL_INTERVAL_MS; - const concurrency = input.concurrency ?? [{ - step: input.step, - active: input.active ?? this.assignment.taskRange.count, - target: input.target ?? this.assignment.taskRange.count, - }]; - this.barrier = { step: input.step, concurrency }; - try { - while (true) { - await this.heartbeat({ currentStep: input.step, concurrency }); - const progress = await this.client.getRunProgress(this.cfg.benchmarkSlug, this.cfg.runId).catch(() => null); - const participant = progress?.participants.find((item) => item.slug === this.cfg.participantSlug); - const step = participant?.concurrency.find((item) => item.step === input.step); - if (step?.ready) { - return { - active: step.active, - target: step.target, - ready: true, - measuredAt: progress?.generatedAt ?? new Date().toISOString(), - }; - } - if (input.timeoutMs !== undefined && Date.now() - startedAt >= input.timeoutMs) { - throw new Error(`Timed out waiting for benchmark step "${input.step}" to become ready.`); - } - await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); - } - } finally { - this.barrier = null; - } - } - - uploadArtifact(input: BenchmarkReporterArtifactInput): Promise { - return this.client.uploadWorkerArtifact(this.cfg.benchmarkSlug, this.cfg.runId, this.assignment.workerId, { - attemptId: this.assignment.attemptId, - kind: input.kind, - name: input.name, - contentType: input.contentType, - metadata: input.metadata, - body: input.body, - }).catch((error) => { - console.warn( - `[benchsdk] failed to upload ${input.kind} artifact for worker ${this.assignment.workerId}: ` + - `${error instanceof Error ? error.message : String(error)}`, - ); - return null; - }); - } - - flush(isFinal = false): Promise { - this.flushChain = this.flushChain.then(async () => { - while (this.pending.length >= this.cfg.batchSize || (isFinal && this.pending.length > 0)) { - const batch = this.pending.slice(0, this.cfg.batchSize); - try { - await this.client.sendTaskResults({ - benchmarkSlug: this.cfg.benchmarkSlug, - runId: this.cfg.runId, - workerId: this.assignment.workerId, - attemptId: this.assignment.attemptId, - sequenceNumber: this.sequenceNumber, - isFinal: isFinal && batch.length === this.pending.length, - records: batch, - }); - } catch (error) { - // Results that never reach the platform are invisible otherwise: the - // worker still completes and the run just reports fewer tasks. - console.warn( - `[benchsdk] dropping ${this.pending.length} unsent task result(s) for worker ${this.assignment.workerId}: ` + - `${error instanceof Error ? error.message : String(error)}`, - ); - break; - } - this.pending.splice(0, batch.length); - this.sequenceNumber += 1; - } - }); - return this.flushChain; - } - - async finish(failed = false, error?: unknown): Promise { - await this.flush(true).catch(() => {}); - if (failed) { - await this.client.failWorker( - this.cfg.benchmarkSlug, - this.cfg.runId, - this.assignment.workerId, - this.assignment.attemptId, - error, - ).catch(() => {}); - return; - } - await this.client.completeWorker( - this.cfg.benchmarkSlug, - this.cfg.runId, - this.assignment.workerId, - this.assignment.attemptId, - ).catch(() => {}); - } -} - -export function claimBenchmarkReporter(config: BenchmarkReporterConfig): Promise { - return BenchmarkReporter.claim(config); -} +export { BenchmarkReporter, claimBenchmarkReporter } from '@benchsdk/worker'; +export type { + BenchmarkReporterArtifactInput, + BenchmarkReporterBarrierInput, + BenchmarkReporterBarrierResult, + BenchmarkReporterConfig, + BenchmarkReporterHeartbeatInput, + BenchmarkReporterProgress, +} from '@benchsdk/worker'; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 79c767b5..c9e6ba6e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -227,6 +227,37 @@ importers: version: 6.0.3 packages/benchsdk: + dependencies: + '@benchsdk/api': + specifier: workspace:* + version: link:../benchsdk-api + '@benchsdk/worker': + specifier: workspace:* + version: link:../benchsdk-worker + devDependencies: + '@types/node': + specifier: ^20.0.0 + version: 20.19.43 + '@vitest/coverage-v8': + specifier: ^1.0.0 + version: 1.6.1(vitest@1.6.1(@types/node@20.19.43)) + eslint: + specifier: ^8.37.0 + version: 8.57.1 + rimraf: + specifier: ^5.0.0 + version: 5.0.10 + tsup: + specifier: ^8.0.0 + version: 8.5.1(jiti@2.7.0)(postcss@8.5.21)(tsx@4.23.12)(typescript@5.9.3)(yaml@2.9.0) + typescript: + specifier: ^5.0.0 + version: 5.9.3 + vitest: + specifier: ^1.0.0 + version: 1.6.1(@types/node@20.19.43) + + packages/benchsdk-api: devDependencies: '@types/node': specifier: ^20.0.0 @@ -272,6 +303,34 @@ importers: specifier: ^1.0.0 version: 1.6.1(@types/node@20.19.43) + packages/benchsdk-worker: + dependencies: + '@benchsdk/api': + specifier: workspace:* + version: link:../benchsdk-api + devDependencies: + '@types/node': + specifier: ^20.0.0 + version: 20.19.43 + '@vitest/coverage-v8': + specifier: ^1.0.0 + version: 1.6.1(vitest@1.6.1(@types/node@20.19.43)) + eslint: + specifier: ^8.37.0 + version: 8.57.1 + rimraf: + specifier: ^5.0.0 + version: 5.0.10 + tsup: + specifier: ^8.0.0 + version: 8.5.1(jiti@2.7.0)(postcss@8.5.21)(tsx@4.23.12)(typescript@5.9.3)(yaml@2.9.0) + typescript: + specifier: ^5.0.0 + version: 5.9.3 + vitest: + specifier: ^1.0.0 + version: 1.6.1(@types/node@20.19.43) + packages/create-bench: devDependencies: '@types/node': diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 4f6b0d00..d603f837 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -1,4 +1,6 @@ packages: + - packages/benchsdk-api + - packages/benchsdk-worker - packages/benchsdk - packages/create-bench - packages/benchsdk-runner From b0ad18505525ca0f13719afa20ab319ae68b9b6c Mon Sep 17 00:00:00 2001 From: david Date: Wed, 19 Aug 2026 00:08:42 +0000 Subject: [PATCH 2/4] ci: build all workspace packages so new @benchsdk/api and @benchsdk/worker artifacts are available Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .github/workflows/ci.yml | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b3ed9f6a..b106448b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,13 +31,9 @@ jobs: run: pnpm install --ignore-scripts - name: Build packages - # The root typecheck consumes the built dist output, so build - # @benchsdk/client, @benchsdk/runner, and create-bench before running - # typecheck. - run: | - pnpm --filter @benchsdk/client run build - pnpm --filter @benchsdk/runner run build - pnpm --filter create-bench run build + # The root typecheck consumes the built dist output, so build all + # workspace packages in dependency order before running typecheck. + run: pnpm -r --filter './packages/**' run build - name: Typecheck run: pnpm typecheck From 55adbf04176a940f7685ac874a5fcf16b37aa3f0 Mon Sep 17 00:00:00 2001 From: david Date: Wed, 19 Aug 2026 00:11:51 +0000 Subject: [PATCH 3/4] fix: pass composed umbrella client to runWorker so WorkerFinishContext.client matches type, and build all packages in release workflow Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .github/workflows/release.yml | 7 +++---- packages/benchsdk/src/client.ts | 10 ++++++---- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 25f278b1..9257d033 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -44,10 +44,9 @@ jobs: run: pnpm install --ignore-scripts - name: Build packages - run: | - pnpm --filter @benchsdk/client run build - pnpm --filter @benchsdk/runner run build - pnpm --filter create-bench run build + # Build all workspace packages in topological order so each package's + # dist output is available before dependents are built. + run: pnpm -r --filter './packages/**' run build - name: Test run: | diff --git a/packages/benchsdk/src/client.ts b/packages/benchsdk/src/client.ts index bac3f54a..b988ad69 100644 --- a/packages/benchsdk/src/client.ts +++ b/packages/benchsdk/src/client.ts @@ -6,10 +6,12 @@ export { BenchmarkApiError }; export function createBenchmarkClient(config: BenchmarkClientConfig = {}): BenchmarkClient { const apiClient = createApiClient(config); - return { + function runWorkerWrapped(options: RunWorkerOptions): Promise { + return runWorker(client as any, options as unknown as any) as Promise; + } + const client = { ...apiClient, - runWorker(options: RunWorkerOptions): Promise { - return runWorker(apiClient, options as unknown as any) as Promise; - }, + runWorker: runWorkerWrapped, } as BenchmarkClient; + return client; } From 43566b34c42c65dfa675a404341553f107d838a8 Mon Sep 17 00:00:00 2001 From: david Date: Wed, 19 Aug 2026 19:03:50 +0000 Subject: [PATCH 4/4] fix(scale): copy and build the new benchsdk workspace packages in the scale image Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- benchmarks/scale/Dockerfile | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/benchmarks/scale/Dockerfile b/benchmarks/scale/Dockerfile index 1d94471e..759b4f0e 100644 --- a/benchmarks/scale/Dockerfile +++ b/benchmarks/scale/Dockerfile @@ -5,12 +5,18 @@ RUN corepack enable && corepack prepare pnpm@10.33.0 --activate WORKDIR /app COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./ COPY packages/benchsdk/package.json packages/benchsdk/ +COPY packages/benchsdk-api/package.json packages/benchsdk-api/ +COPY packages/benchsdk-worker/package.json packages/benchsdk-worker/ COPY tsconfig.json ./ RUN pnpm install --frozen-lockfile --ignore-scripts COPY benchmarks ./benchmarks COPY packages/benchsdk ./packages/benchsdk -RUN pnpm --filter @benchsdk/client run build +COPY packages/benchsdk-api ./packages/benchsdk-api +COPY packages/benchsdk-worker ./packages/benchsdk-worker +# `...` builds @benchsdk/client's workspace dependencies first, so the api and +# worker packages exist before the client's tsup/dts build resolves them. +RUN pnpm --filter "@benchsdk/client..." run build RUN pnpm run --silent bundle:scale FROM node:24-alpine