diff --git a/docs/benchmarks/enrichment-baseline-2026-10-14.json b/docs/benchmarks/enrichment-baseline-2026-10-14.json new file mode 100644 index 00000000..2bbc6dc6 --- /dev/null +++ b/docs/benchmarks/enrichment-baseline-2026-10-14.json @@ -0,0 +1,1066 @@ +{ + "note": "Developer-laptop M0 baseline; regression reference only, not capacity claims.", + "runs": [ + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=500 fixed 50ms defaults", + "ok": true, + "started_at": "2026-10-13T18:16:49.404116+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": false + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 500, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 500, + "completed": 500, + "statuses": { + "complete": 500 + }, + "errors": {}, + "overlay_rows": 500, + "attempts_recorded": 500 + }, + "throughput": { + "completed_per_second": 70.14, + "completed_per_second_excluding_startup": 77.23 + }, + "latency_ms": { + "end_to_end": { + "count": 500, + "p50": 4227.2, + "p95": 6852.4, + "p99": 7121.2, + "max": 7128.4 + }, + "cell_execution": { + "count": 500, + "p50": 87.1, + "p95": 132.2, + "p99": 701.5, + "max": 725.1 + }, + "startup_to_first_cell": 653.8 + }, + "wall_seconds": { + "enqueue_to_last_cell": 7.128, + "enqueue_to_jobs_done": 7.476 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 500, + "unique_keys": 500, + "retried_keys": 0, + "status_counts": { + "200": 500 + }, + "peak_in_flight": 8, + "uptime_seconds": 8.682 + }, + "resources": { + "harness_cpu_seconds": 1.12, + "children_cpu_seconds": 3.46, + "harness_peak_rss_mb": 170.5, + "largest_child_peak_rss_mb": 160.6, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=2000 fixed 50ms defaults", + "ok": true, + "started_at": "2026-10-13T18:17:11.456626+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": false + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 2000, + "statuses": { + "complete": 2000 + }, + "errors": {}, + "overlay_rows": 2000, + "attempts_recorded": 2000 + }, + "throughput": { + "completed_per_second": 80.62, + "completed_per_second_excluding_startup": 82.9 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 13248.3, + "p95": 23697.0, + "p99": 24587.4, + "max": 24807.5 + }, + "cell_execution": { + "count": 2000, + "p50": 85.3, + "p95": 129.6, + "p99": 138.9, + "max": 803.2 + }, + "startup_to_first_cell": 680.8 + }, + "wall_seconds": { + "enqueue_to_last_cell": 24.807, + "enqueue_to_jobs_done": 25.23 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 2000, + "unique_keys": 2000, + "retried_keys": 0, + "status_counts": { + "200": 2000 + }, + "peak_in_flight": 8, + "uptime_seconds": 26.247 + }, + "resources": { + "harness_cpu_seconds": 1.17, + "children_cpu_seconds": 10.65, + "harness_peak_rss_mb": 172.2, + "largest_child_peak_rss_mb": 162.6, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=2000 zero latency defaults (overhead ceiling)", + "ok": true, + "started_at": "2026-10-13T18:17:38.290665+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": false + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 0.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 2000, + "statuses": { + "complete": 2000 + }, + "errors": {}, + "overlay_rows": 2000, + "attempts_recorded": 2000 + }, + "throughput": { + "completed_per_second": 214.58, + "completed_per_second_excluding_startup": 230.53 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 5346.8, + "p95": 8950.1, + "p99": 9243.0, + "max": 9320.7 + }, + "cell_execution": { + "count": 2000, + "p50": 40.4, + "p95": 47.4, + "p99": 93.8, + "max": 669.6 + }, + "startup_to_first_cell": 645.1 + }, + "wall_seconds": { + "enqueue_to_last_cell": 9.321, + "enqueue_to_jobs_done": 9.692 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 2000, + "unique_keys": 2000, + "retried_keys": 0, + "status_counts": { + "200": 2000 + }, + "peak_in_flight": 1, + "uptime_seconds": 10.86 + }, + "resources": { + "harness_cpu_seconds": 1.1, + "children_cpu_seconds": 9.93, + "harness_peak_rss_mb": 172.1, + "largest_child_peak_rss_mb": 163.0, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=2000 lognormal 50ms sigma0.5 1% 5xx 2% 429 defaults", + "ok": true, + "started_at": "2026-10-13T18:17:56.076757+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": false + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "lognormal", + "jitter_ms": 0.0, + "sigma": 0.5, + "error_rate": 0.01, + "rate_429": 0.02 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 1999, + "statuses": { + "complete": 1999, + "error": 1 + }, + "errors": { + "no_data": 1 + }, + "overlay_rows": 2000, + "attempts_recorded": 2061 + }, + "throughput": { + "completed_per_second": 61.7, + "completed_per_second_excluding_startup": 62.96 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 16487.2, + "p95": 30489.8, + "p99": 32078.1, + "max": 32401.2 + }, + "cell_execution": { + "count": 2061, + "p50": 93.3, + "p95": 169.8, + "p99": 234.4, + "max": 677.4 + }, + "startup_to_first_cell": 649.9 + }, + "wall_seconds": { + "enqueue_to_last_cell": 32.401, + "enqueue_to_jobs_done": 32.873 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 2061, + "unique_keys": 2000, + "retried_keys": 61, + "status_counts": { + "200": 1999, + "429": 38, + "500": 24 + }, + "peak_in_flight": 8, + "uptime_seconds": 33.852 + }, + "resources": { + "harness_cpu_seconds": 1.19, + "children_cpu_seconds": 11.61, + "harness_peak_rss_mb": 172.0, + "largest_child_peak_rss_mb": 163.0, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=2000 fixed 50ms rows12 pw12", + "ok": true, + "started_at": "2026-10-13T18:39:07.360716+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": true + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 12, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 2000, + "statuses": { + "complete": 2000 + }, + "errors": {}, + "overlay_rows": 2000, + "attempts_recorded": 2000 + }, + "throughput": { + "completed_per_second": 112.92, + "completed_per_second_excluding_startup": 117.32 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 9825.9, + "p95": 16947.9, + "p99": 17532.3, + "max": 17710.9 + }, + "cell_execution": { + "count": 2000, + "p50": 84.3, + "p95": 94.0, + "p99": 133.2, + "max": 960.3 + }, + "startup_to_first_cell": 662.9 + }, + "wall_seconds": { + "enqueue_to_last_cell": 17.711, + "enqueue_to_jobs_done": 18.03 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 2000, + "unique_keys": 2000, + "retried_keys": 0, + "status_counts": { + "200": 2000 + }, + "peak_in_flight": 12, + "uptime_seconds": 19.424 + }, + "resources": { + "harness_cpu_seconds": 1.16, + "children_cpu_seconds": 9.89, + "harness_peak_rss_mb": 171.6, + "largest_child_peak_rss_mb": 161.9, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "N=2000 fixed 50ms 4 workbooks/2 workspaces slots4", + "ok": true, + "started_at": "2026-10-13T18:39:27.383209+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": true + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "sqlite", + "version": "3.50.4", + "location": "temporary file" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 4, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 4, + "workspaces": 2, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 4, + "statuses": { + "completed": 4 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 2000, + "statuses": { + "complete": 2000 + }, + "errors": {}, + "overlay_rows": 2000, + "attempts_recorded": 2000 + }, + "throughput": { + "completed_per_second": 189.23, + "completed_per_second_excluding_startup": 203.7 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 6882.2, + "p95": 10063.1, + "p99": 10389.1, + "max": 10569.3 + }, + "cell_execution": { + "count": 2000, + "p50": 122.2, + "p95": 194.9, + "p99": 2481.9, + "max": 2527.5 + }, + "startup_to_first_cell": 751.0 + }, + "wall_seconds": { + "enqueue_to_last_cell": 10.569, + "enqueue_to_jobs_done": 10.976 + }, + "dialect": "sqlite" + }, + "simulator_stats": { + "requests": 2000, + "unique_keys": 2000, + "retried_keys": 0, + "status_counts": { + "200": 2000 + }, + "peak_in_flight": 32, + "uptime_seconds": 12.141 + }, + "resources": { + "harness_cpu_seconds": 1.15, + "children_cpu_seconds": 14.88, + "harness_peak_rss_mb": 172.5, + "largest_child_peak_rss_mb": 161.0, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "PG N=500", + "ok": true, + "started_at": "2026-10-13T18:39:47.061977+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": true + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "postgresql", + "version": "18.4 (Homebrew)", + "location": "caller-supplied" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 500, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 500, + "completed": 500, + "statuses": { + "complete": 500 + }, + "errors": {}, + "overlay_rows": 500, + "attempts_recorded": 500 + }, + "throughput": { + "completed_per_second": 45.05, + "completed_per_second_excluding_startup": 48.06 + }, + "latency_ms": { + "end_to_end": { + "count": 500, + "p50": 6204.0, + "p95": 10671.7, + "p99": 11098.6, + "max": 11099.2 + }, + "cell_execution": { + "count": 500, + "p50": 183.4, + "p95": 219.0, + "p99": 661.9, + "max": 721.9 + }, + "startup_to_first_cell": 695.2 + }, + "wall_seconds": { + "enqueue_to_last_cell": 11.099, + "enqueue_to_jobs_done": 11.415 + }, + "dialect": "postgresql" + }, + "simulator_stats": { + "requests": 500, + "unique_keys": 500, + "retried_keys": 0, + "status_counts": { + "200": 500 + }, + "peak_in_flight": 8, + "uptime_seconds": 17.188 + }, + "resources": { + "harness_cpu_seconds": 1.53, + "children_cpu_seconds": 4.75, + "harness_peak_rss_mb": 188.8, + "largest_child_peak_rss_mb": 167.7, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + }, + { + "schema_version": 1, + "benchmark": "enrichment_completion", + "label": "PG N=2000", + "ok": true, + "started_at": "2026-10-13T18:40:09.865994+00:00", + "identity": { + "git": { + "commit": "9188fc4d040c186f3495902fe77e0d041deba3bd", + "branch": "worktree-agent-aa4aceef986fd2e13", + "dirty": true + }, + "hardware": { + "cpu_model": "Apple M2", + "physical_cores": 8, + "logical_cores": 8, + "ram_gb": 16.0, + "os": "macOS-26.4.1-arm64-arm-64bit-Mach-O", + "machine": "arm64" + }, + "python": { + "version": "3.14.5", + "implementation": "CPython" + }, + "database": { + "dialect": "postgresql", + "version": "18.4 (Homebrew)", + "location": "caller-supplied" + }, + "redis": { + "reachable": true + }, + "worker_config": { + "worker_concurrency": 1, + "max_active_per_workspace": 2, + "worker_processes": 1, + "heartbeat_interval_seconds": 30, + "run_workbook_timeout_seconds": 1800, + "row_concurrency": 12, + "provider_workers": 8, + "provider_timeout_seconds": 10, + "max_providers": 8, + "retry_passes": 1 + }, + "workload": { + "rows": 2000, + "workbooks": 1, + "workspaces": 1, + "columns": [ + { + "type": "enrichment", + "provider": "bench_simulator", + "path": "declarative manifest -> pebble provider worker -> httpx" + } + ], + "cost_per_lookup": 0.0, + "simulator": { + "seed": 42, + "latency_ms": 50.0, + "latency_dist": "fixed", + "jitter_ms": 0.0, + "sigma": 0.25, + "error_rate": 0.0, + "rate_429": 0.0 + } + } + }, + "results": { + "timed_out": false, + "jobs": { + "count": 1, + "statuses": { + "completed": 1 + }, + "retries": 0, + "errors": [] + }, + "cells": { + "total": 2000, + "completed": 2000, + "statuses": { + "complete": 2000 + }, + "errors": {}, + "overlay_rows": 2000, + "attempts_recorded": 2000 + }, + "throughput": { + "completed_per_second": 37.79, + "completed_per_second_excluding_startup": 38.32 + }, + "latency_ms": { + "end_to_end": { + "count": 2000, + "p50": 20794.4, + "p95": 49864.8, + "p99": 52167.2, + "max": 52925.5 + }, + "cell_execution": { + "count": 2000, + "p50": 214.2, + "p95": 410.8, + "p99": 469.1, + "max": 904.2 + }, + "startup_to_first_cell": 735.6 + }, + "wall_seconds": { + "enqueue_to_last_cell": 52.926, + "enqueue_to_jobs_done": 53.276 + }, + "dialect": "postgresql" + }, + "simulator_stats": { + "requests": 2000, + "unique_keys": 2000, + "retried_keys": 0, + "status_counts": { + "200": 2000 + }, + "peak_in_flight": 8, + "uptime_seconds": 54.341 + }, + "resources": { + "harness_cpu_seconds": 1.33, + "children_cpu_seconds": 21.01, + "harness_peak_rss_mb": 174.3, + "largest_child_peak_rss_mb": 168.2, + "note": "children = job child + provider workers + simulator (reaped processes)" + } + } + ] +} diff --git a/docs/benchmarks/enrichment-baseline.md b/docs/benchmarks/enrichment-baseline.md new file mode 100644 index 00000000..bce88601 --- /dev/null +++ b/docs/benchmarks/enrichment-baseline.md @@ -0,0 +1,92 @@ +# Enrichment completion baseline (M0) + +**Developer-laptop baseline.** These numbers come from one development machine +with simulated providers. Use them as a regression reference only. They are +not capacity claims and do not meet the M0 exit gate in +[the backend plan](../plans/go-python-backend-rewrite.md), which needs a +controlled environment, PostgreSQL, steady-state duration, and API +measurements. Raw reports: +[enrichment-baseline-2026-10-14.json](enrichment-baseline-2026-10-14.json). +For how to run the harness and what each knob does, see +[operations/worker-concurrency.md](../operations/worker-concurrency.md). + +## Identity + +| Field | Value | +| --- | --- | +| Commit | `9188fc4d040c186f3495902fe77e0d041deba3bd` (application code; harness files added on top, uncommitted at run time) | +| Hardware | Apple M2, 8 cores (8 logical), 16 GB RAM | +| OS | macOS 26.4.1 (arm64) | +| Python | CPython 3.14.5 (uv-managed venv) | +| Database | SQLite 3.50.4 (temporary file, WAL); PostgreSQL 18.4 (Homebrew, local socket, superuser role, so RLS policies were bypassed) | +| Redis | Reachable on localhost (cell broadcasts enabled) | +| Worker processes | 1 harness worker process (`QueueService`), 1 job child per slot | +| Defaults used unless noted | `WORKER_CONCURRENCY=1`, `WORKER_MAX_ACTIVE_PER_WORKSPACE=2`, `row_concurrency=12`, `provider_workers=8`, `provider_timeout=10`, `max_providers=8`, `retry_passes=1`, heartbeat 30 s, `run_workbook` timeout 1800 s | +| Workload | N rows x 1 `enrichment` column on a declarative HTTP provider (manifest -> pebble provider worker -> httpx -> simulator), `verify: false`, `cost_per_lookup: 0`, one provider call per cell | +| Simulator | Seed 42; latency, distribution, and error rates per row below | +| Date | 2026-10-14 | + +## Results + +Throughput is committed completions/sec from enqueue to the last result +commit; the number in parentheses excludes the job-start overhead. "E2E" +means enqueue to the cell's result commit (one batch enqueue, so it includes +queueing). "Cell" is one `enrich_cell` execution: provider call plus commit. + +| Run | DB | Rows | Simulator | Config change | Done | Completions/s | E2E p50 / p95 / p99 (ms) | Cell p50 / p95 / p99 (ms) | Wall (s) | +| --- | --- | ---: | --- | --- | ---: | ---: | --- | --- | ---: | +| A | SQLite | 500 | fixed 50 ms | defaults | 500 | 70.1 (77.2) | 4227 / 6852 / 7121 | 87 / 132 / 702 | 7.5 | +| B | SQLite | 2000 | fixed 50 ms | defaults | 2000 | 80.6 (82.9) | 13248 / 23697 / 24587 | 85 / 130 / 139 | 25.2 | +| C | SQLite | 2000 | fixed 0 ms | defaults | 2000 | 214.6 (230.5) | 5347 / 8950 / 9243 | 40 / 47 / 94 | 9.7 | +| D | SQLite | 2000 | lognormal 50 ms, sigma 0.5, 1% 5xx, 2% 429 | defaults | 1999 | 61.7 (63.0) | 16487 / 30490 / 32078 | 93 / 170 / 234 | 32.9 | +| E | SQLite | 2000 | fixed 50 ms | `provider_workers=12` | 2000 | 112.9 (117.3) | 9826 / 16948 / 17532 | 84 / 94 / 133 | 18.0 | +| F | SQLite | 2000 | fixed 50 ms | 4 workbooks / 2 workspaces, `WORKER_CONCURRENCY=4` | 2000 | 189.2 (203.7) | 6882 / 10063 / 10389 | 122 / 195 / 2482 | 11.0 | +| G | PostgreSQL | 500 | fixed 50 ms | defaults | 500 | 45.1 (48.1) | 6204 / 10672 / 11099 | 183 / 219 / 662 | 11.4 | +| H | PostgreSQL | 2000 | fixed 50 ms | defaults | 2000 | 37.8 (38.3) | 20794 / 49865 / 52167 | 214 / 411 / 469 | 53.3 | + +Provider attempts equal rows in every run except D, which made 2061 requests: +1999 returned 200, 38 returned 429, and 24 returned 5xx. The retry pass +recovered all but one cell. Peak simulator in-flight requests matched +`provider_workers` per job (8, 12, or 32 = 4 jobs x 8). Job startup (claim, +child process, provider registry, pool start) took 0.65-0.75 s to the first +cell. Peak RSS per process was about 160-170 MB. + +## Observations to carry into M0/M3 + +1. **Provider pool and batching set throughput.** With the defaults, 12 rows + per batch share 8 provider processes, and each batch waits for its slowest + row. Raising `provider_workers` to match `row_concurrency` gave +40% + (B -> E). The zero-latency ceiling (C) is about 215/s per job on SQLite. +2. **Row concurrency above the DB pool deadlocks a run.** With + `row_concurrency=32` (allowed by Settings, up to 64), the job child blocked + its event loop on a SQLAlchemy `QueuePool` checkout ("limit of size 5 + overflow 10 reached"). A 300-row run finished 0 cells in 240 s, and an + earlier 2000-row attempt finished 14 cells in about 10 minutes. This + should be a bug fix (bound or decouple sessions from in-flight provider + calls), not a tuning note. +3. **PostgreSQL is slower than SQLite here and degrades with N.** Cell p95 went + from 219 ms (G, 500 rows) to 411 ms (H, 2000 rows) at the same provider + latency. Per-cell database work appears to grow with workbook size. Profile + this before M3 ("bounded workbook refresh queries"). +4. **Failure detail is lost.** The cell that exhausted retries in D recorded + `no_data`. The provider's `http_429`/`http_500` error does not reach the + cell, so 429s are not visible per cell and do not count as rate-limit + signals. +5. **Slots scale across workbooks, not within one.** Four slots and four + workbooks reached 189/s (F). A single large workbook uses one slot no + matter what `WORKER_CONCURRENCY` is set to. + +## Reproduce + +```bash +uv run python scripts/bench/enrichment_harness.py --rows 500 --latency-ms 50 --output a.json +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 --output b.json +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 0 --output c.json +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 --latency-dist lognormal \ + --sigma 0.5 --error-rate 0.01 --rate-429 0.02 --output d.json +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 --provider-workers 12 --output e.json +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 --workbooks 4 --workspaces 2 \ + --worker-concurrency 4 --output f.json +BENCH_DATABASE_URL='postgresql+psycopg://USER@/opengtm_bench?host=/tmp' \ + uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 --output h.json +``` diff --git a/docs/operations/worker-concurrency.md b/docs/operations/worker-concurrency.md new file mode 100644 index 00000000..7adac09c --- /dev/null +++ b/docs/operations/worker-concurrency.md @@ -0,0 +1,147 @@ +# Worker slots and enrichment concurrency + +Workbook enrichment concurrency is bounded at four nested levels. Each level +caps a different resource, and the effective parallelism is the *minimum* +along the path, not the product. This page lists every knob, its default, what +it bounds, and how to tune it. Measure changes with the +[enrichment completion harness](#measuring-with-the-enrichment-harness). + +``` +replicas x WORKER_CONCURRENCY slots (durable jobs at once) + └─ capped per tenant by WORKER_MAX_ACTIVE_PER_WORKSPACE + └─ one killable job child per slot (run_workbook) + └─ row_concurrency rows per batch (asyncio, one DB session per cell) + └─ provider_workers pebble processes (killable provider calls) +``` + +## Durable queue (per worker process) + +| Setting | Default | Range | Bounds | +| --- | --- | --- | --- | +| `WORKER_CONCURRENCY` | `1` | 1-64 | Claim slots per worker process (the in-API worker or each `python -m apps.api.worker` replica). Each slot claims one job and runs it in its own killable child process (`apps.api.job_process`). Total job parallelism is `replicas x WORKER_CONCURRENCY`. | +| `WORKER_MAX_ACTIVE_PER_WORKSPACE` | `2` | 0-64 (0 disables) | Jobs one workspace may have in `processing` at once, across all replicas. PostgreSQL enforces it with a per-workspace advisory lock during the claim; global jobs (no workspace) are uncapped. This is the tenant-fairness bound. | +| `WORKER_SHUTDOWN_GRACE_SECONDS` | `30` | 0-300 | How long shutdown drains active slots before cancelling (and killing) their job children. | +| `RUN_INLINE_WORKER` | `1` | `0`/`1` | Whether the API process also runs a worker. Set `0` when standalone worker replicas own processing. | +| `run_workbook` timeout | 1800 s | code (`JOB_TIMEOUTS`) | Wall-clock limit on one enrichment job; the child is killed on expiry. | +| Heartbeat interval | 30 s | code | Lease renewal; dead-worker reaping uses it. | + +A single `run_workbook` job occupies one slot for its whole run, so raising +`WORKER_CONCURRENCY` speeds up *several* workbooks (or tenants) running at once, +not one large workbook. With the default of one slot, a second workbook waits +until the first finishes. + +## Per-run enrichment settings (Settings -> Enrichment performance) + +These are stored in the settings database as `ENRICH_` (set them in the UI +or with `PUT /api/settings/enrichment-perf`, admin only). The API snapshots them +into each `run_workbook` job payload at enqueue time, so a change affects the +next run, not one already queued. + +| Setting | Default | Range | Bounds | +| --- | --- | --- | --- | +| `row_concurrency` | `12` | 1-64 | Rows enriched concurrently inside one job. Rows run in batches of this size and each batch waits for its slowest row before the next starts. Each in-flight cell holds one database session. | +| `provider_workers` | `8` | 1-64 | Size of the pebble process pool for provider calls inside one job child. Caps concurrent provider calls per job; a hung call costs one killed process, not the run. | +| `provider_timeout` | `10` s | 3-120 | Per provider call; the provider process is killed when it expires. | +| `max_providers` | `8` | 0-20 (0 = full chain) | Waterfall depth per cell. | +| `retry_passes` | `1` | 0-5 | Extra passes over cells still in `error` after the main pass. | + +Environment fallbacks used when a payload or the settings database lacks a +value: `WORKBOOK_RUN_CONCURRENCY` (12), `WORKBOOK_PROVIDER_WORKERS` (8), +`WORKBOOK_PROVIDER_TIMEOUT` (10), `WORKBOOK_MAX_PROVIDERS_PER_CELL` (0). + +## Database connection pools (per process) + +Every job child creates its own SQLAlchemy engine: + +| Backend | Pool | +| --- | --- | +| SQLite | SQLAlchemy default `QueuePool`: 5 connections + 10 overflow, 30 s checkout timeout | +| PostgreSQL | `pool_size=10`, `max_overflow=20` (30 connections), 30 s checkout timeout | + +PostgreSQL `max_connections` must cover +`replicas x (1 + WORKER_CONCURRENCY x per-child pool)` plus the API. + +## Tuning + +1. **Keep `row_concurrency` below the job child's pool capacity**: at most 12 + on SQLite (15 connections) and about 25 on PostgreSQL (30 connections). + Cells keep their session open across the provider call, and a checkout + that has to wait blocks the event loop of the job child. Every in-flight + cell then stalls until the 30 s pool timeout fires. The M0 harness + reproduced this: `row_concurrency=32` on SQLite finished 14 cells in about + ten minutes. The Settings UI currently allows up to 64. Track this as a + bug; do not tune around it. +2. **Match `provider_workers` to `row_concurrency`** for single-provider + columns. A provider call is the only work that uses the pool, so with + `provider_workers < row_concurrency` part of each batch queues behind the + rest. Each provider worker is a full Python process (about 100-160 MB RSS), + so per host memory is roughly + `replicas x WORKER_CONCURRENCY x provider_workers` processes. +3. **Raise `WORKER_CONCURRENCY`** when several workbooks or tenants run at + once. Keep `WORKER_MAX_ACTIVE_PER_WORKSPACE` below it so one tenant cannot + take every slot. +4. **Use PostgreSQL** for concurrent writers. SQLite serializes writes and is a + development backend. +5. Re-measure after every change with the same harness parameters and record + the identity block next to the result. + +## Measuring with the enrichment harness + +`scripts/bench/enrichment_harness.py` runs the real path: durable queue claim, +`run_workbook` job child, declarative HTTP provider in the pebble pool, and +result commit. Calls go to a deterministic local provider simulator +(`scripts/bench/provider_simulator.py`). + +```bash +# Shipped defaults, 500 rows, fixed 50 ms provider latency, temp SQLite +uv run python scripts/bench/enrichment_harness.py --rows 500 --latency-ms 50 \ + --output artifacts/bench/enrichment-500.json + +# Fault mix: lognormal latency, 1% 5xx, 2% 429, fixed seed +uv run python scripts/bench/enrichment_harness.py --rows 2000 --latency-ms 50 \ + --latency-dist lognormal --sigma 0.5 --error-rate 0.01 --rate-429 0.02 --seed 42 + +# Override any knob (defaults are the shipped values above) +uv run python scripts/bench/enrichment_harness.py --rows 2000 --workbooks 4 \ + --workspaces 2 --worker-concurrency 4 --row-concurrency 12 --provider-workers 12 + +# PostgreSQL: a disposable, migrated database only +BENCH_DATABASE_URL=postgresql+psycopg://.../opengtm_bench \ + uv run python scripts/bench/enrichment_harness.py --rows 2000 +``` + +The report (`--output` writes JSON, `--json` prints it) includes: + +- **Results:** committed completions/sec (from enqueue, and excluding the + job-start overhead); end-to-end latency per cell (enqueue to result commit) + and cell execution latency at p50/p95/p99; startup time to the first cell; + cell status and error counts; job statuses; simulator request and status + counts (provider attempts, 429s, 5xx); approximate CPU and peak RSS. +- **Identity block:** git commit and dirty flag; CPU model, cores, RAM, and OS; + Python version; database dialect and version; Redis reachability; every + worker and enrichment setting above; workload parameters (rows, workbooks, + workspaces, simulator latency, error rates, and seed). + +How it works and what it changes: + +- Simulator outcomes are a pure function of `(seed, row domain, attempt)`, so + identical parameters produce the same provider behavior regardless of + scheduling. +- Benchmark processes load `scripts/bench/_shim/sitecustomize.py` through + `PYTHONPATH`. It makes three changes: it exempts exactly the simulator's + loopback origin from the SSRF guard (any other URL is still checked), it + registers the generated manifest through the normal compile/register path, + and it records per-cell timings. It refuses any non-loopback origin. None of + this ships in the application. +- With the default temporary SQLite database, nothing persists. With a + supplied database, the harness refuses to start if other jobs are + pending/processing. It claims only its own fire-key-tagged jobs and deletes + only the rows it created. +- The email verify cascade is disabled on the benchmark column (it would + measure live DNS/SMTP). Pass `--cost-per-lookup` > 0 to route calls through + the spend-reservation path. +- `kill -USR1 ` dumps all thread stacks of a stalled benchmark + process to stderr. + +Developer-laptop numbers are a regression reference only. They are not capacity +claims; see [benchmarks/enrichment-baseline.md](../benchmarks/enrichment-baseline.md). diff --git a/docs/plans/go-python-backend-rewrite.md b/docs/plans/go-python-backend-rewrite.md index 9b99659e..85538000 100644 --- a/docs/plans/go-python-backend-rewrite.md +++ b/docs/plans/go-python-backend-rewrite.md @@ -164,7 +164,7 @@ process termination and parent-side failure reconciliation. | Milestone | Deliverable | Exit gate | | --- | --- | --- | -| M0: baseline | Fixed fixtures, provider simulator, representative workloads, resource profiles, and current queue/API measurements | Reproducible baseline with hardware, commit, configuration, and workload identity | +| M0: baseline | Fixed fixtures, provider simulator, representative workloads, resource profiles, and current queue/API measurements ([harness and settings](../operations/worker-concurrency.md), [developer-laptop baseline](../benchmarks/enrichment-baseline.md)) | Reproducible baseline with hardware, commit, configuration, and workload identity | | M1: foundations | Go module, sqlc queries, versioned contracts, telemetry, additive executor routing/fencing, and Compose development setup | Python-only operation still passes; mixed-version claims, RLS, and lease recovery tests pass | | M2: one worker | One declarative HTTP enrichment job type in Go, including result/evidence persistence, billing coordination, and progress | Behavioral parity, crash/cancellation tests, benchmark comparison, feature flag, and exercised rollback | | M3: execution improvements | Bounded concurrency, transport pooling, shared limits, batching, and bounded workbook refresh queries | Measured improvement without degraded fairness, correctness, or resource bounds | @@ -241,6 +241,10 @@ PostgreSQL correctness, or reproducible performance testing. Follow [CONTRIBUTING.md](../../CONTRIBUTING.md), including DCO sign-off. - [ ] Record M0 baseline and select the first enrichment job type. + - Harness, simulator, and a first developer-laptop baseline are in + [benchmarks/enrichment-baseline.md](../benchmarks/enrichment-baseline.md). + Still needed: a controlled-environment PostgreSQL baseline, API (k6) + measurements, and the first job type selection. - [ ] Define domain ownership, task contracts, executor routing, and lease fencing. - [ ] Add the Go development skeleton, pgx/sqlc access, and tenant transaction helper. - [ ] Preserve Python-only operation and exercise mixed-executor claims. diff --git a/scripts/bench/_shim/sitecustomize.py b/scripts/bench/_shim/sitecustomize.py new file mode 100644 index 00000000..5be13bde --- /dev/null +++ b/scripts/bench/_shim/sitecustomize.py @@ -0,0 +1,163 @@ +"""Benchmark-only process hooks for the enrichment completion harness. + +The harness puts this directory on ``PYTHONPATH`` so every Python process it +starts (the durable job child and the pebble provider workers forked from the +forkserver) imports this module at interpreter startup. It does nothing unless +``OPENGTM_BENCH_SIMULATOR_ORIGIN`` is set, and it never ships with the app. + +When active it installs post-import patches (applied as each target module is +first executed, so ``from x import y`` callers see the patched object): + +* ``apps.api.core.url_guard.check_url`` allows exactly one loopback origin, the + provider simulator (scheme + host + port must match). Every other URL still + goes through the unmodified SSRF guard. +* ``register_declarative_manifests`` also registers the harness-generated + manifest at ``OPENGTM_BENCH_MANIFEST`` through the normal compile/register + path, so the run uses the real ``DeclarativeProvider``. +* ``enrich_cell`` is wrapped to append one JSON line per cell attempt (wall-clock + start/end, outcome) to ``OPENGTM_BENCH_TIMINGS`` for latency percentiles. +""" + +from __future__ import annotations + +import importlib.abc +import json +import os +import sys +import time +from urllib.parse import urlparse + +_ORIGIN = os.environ.get("OPENGTM_BENCH_SIMULATOR_ORIGIN", "").rstrip("/") + + +def _origin_key(url: str) -> tuple[str, str, int | None]: + parsed = urlparse(url) + port = parsed.port or {"http": 80, "https": 443}.get(parsed.scheme) + return parsed.scheme, (parsed.hostname or "").lower(), port + + +def _is_loopback_origin(origin: str) -> bool: + import ipaddress + + scheme, host, port = _origin_key(origin) + try: + return scheme == "http" and port is not None and ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def _patch_url_guard(module) -> None: + original = module.check_url + allowed = _origin_key(_ORIGIN) + if not _is_loopback_origin(_ORIGIN): + # Only a loopback simulator may be exempted; never widen egress to a + # private-network or metadata host, even in benchmark processes. + raise RuntimeError("OPENGTM_BENCH_SIMULATOR_ORIGIN must be http://:") + + def check_url(url, *args, **kwargs): + if isinstance(url, str) and _origin_key(url.strip()) == allowed: + return url + return original(url, *args, **kwargs) + + check_url.__wrapped__ = original # type: ignore[attr-defined] + module.check_url = check_url + + +def _patch_declarative_registry(module) -> None: + original = module.register_declarative_manifests + manifest_path = os.environ.get("OPENGTM_BENCH_MANIFEST", "") + + def register_declarative_manifests() -> int: + count = original() + if manifest_path: + from pathlib import Path + + from apps.api.services.leadgen.enrichment.declarative.compiler import ( + compile_manifest, + ) + from apps.api.services.leadgen.enrichment.declarative.manifest import ( + load_manifest, + ) + from apps.api.services.workbook.providers import register_provider + + register_provider(compile_manifest(load_manifest(Path(manifest_path)))) + count += 1 + return count + + module.register_declarative_manifests = register_declarative_manifests + + +def _patch_enrichment(module) -> None: + timings_path = os.environ.get("OPENGTM_BENCH_TIMINGS", "") + if not timings_path: + return + original = module.enrich_cell + + async def enrich_cell(*args, **kwargs): + started = time.time() + result = None + try: + result = await original(*args, **kwargs) + return result + finally: + lead_data = kwargs.get("lead_data") or {} + record = { + "row_id": lead_data.get("__row_id"), + "workbook_id": kwargs.get("workbook_id"), + "column_id": kwargs.get("col_id"), + "start": started, + "end": time.time(), + "success": bool(isinstance(result, dict) and result.get("success")), + "error": (result or {}).get("error") if isinstance(result, dict) else "exception", + } + # One short O_APPEND write per line; safe across the job children. + with open(timings_path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, separators=(",", ":")) + "\n") + + module.enrich_cell = enrich_cell + + +_PATCHES = { + "apps.api.core.url_guard": _patch_url_guard, + "apps.api.services.leadgen.enrichment.declarative.registry": _patch_declarative_registry, + "apps.api.services.workbook.enrichment": _patch_enrichment, +} + + +class _PatchingLoader(importlib.abc.Loader): + def __init__(self, inner, patch): + self._inner = inner + self._patch = patch + + def create_module(self, spec): + return self._inner.create_module(spec) + + def exec_module(self, module): + self._inner.exec_module(module) + self._patch(module) + + +class _PatchingFinder(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path, target=None): + patch = _PATCHES.get(fullname) + if patch is None: + return None + for finder in sys.meta_path: + if finder is self or not hasattr(finder, "find_spec"): + continue + spec = finder.find_spec(fullname, path, target) + if spec is not None and spec.loader is not None: + spec.loader = _PatchingLoader(spec.loader, patch) + return spec + return None + + +if _ORIGIN and not any(isinstance(f, _PatchingFinder) for f in sys.meta_path): + sys.meta_path.insert(0, _PatchingFinder()) + # `kill -USR1 ` dumps every thread's stack to stderr: diagnoses stalls + # without root (py-spy needs root on macOS). + import faulthandler + import signal + + if hasattr(signal, "SIGUSR1"): + faulthandler.register(signal.SIGUSR1, all_threads=True) diff --git a/scripts/bench/enrichment_harness.py b/scripts/bench/enrichment_harness.py new file mode 100755 index 00000000..fd427e01 --- /dev/null +++ b/scripts/bench/enrichment_harness.py @@ -0,0 +1,586 @@ +#!/usr/bin/env python3 +"""Enrichment completion harness (M0 baseline, issue #33). + +Measures how fast the *current* Python stack completes workbook enrichments +end to end, against the deterministic provider simulator: + +1. starts ``provider_simulator.py`` on an ephemeral loopback port; +2. writes a declarative provider manifest pointing at it (the real + ``DeclarativeProvider`` compile/register path, run in the killable pebble + provider pool); +3. creates N workbook rows with one ``enrichment`` column using that provider; +4. enqueues ``run_workbook`` jobs on the durable SQL queue and runs a real + ``QueueService`` worker (claim -> job child process -> provider workers -> + result commit); +5. waits for every job to finish and reports committed completions/sec, + per-cell latency percentiles, errors, provider attempts/429s, CPU/RSS, and + the identity block (commit, hardware, Python, DB, worker config, workload). + + uv run python scripts/bench/enrichment_harness.py --rows 500 --latency-ms 50 \ + --output artifacts/bench/enrichment-500.json + +SQLite in a temporary directory is the default. Set ``--database-url`` (or +``BENCH_DATABASE_URL``) to a *disposable*, migrated PostgreSQL database to +measure Postgres; the harness only claims and deletes rows it created. + +Benchmark-only tooling. The SSRF guard is exempted for exactly the simulator's +loopback origin inside benchmark processes (see ``_shim/sitecustomize.py``). +""" + +from __future__ import annotations + +import argparse +import asyncio +import functools +import json +import math +import os +import platform +import resource +import shutil +import subprocess +import sys +import tempfile +import time +import urllib.request +import uuid +from collections import Counter +from datetime import UTC, datetime, timezone +from pathlib import Path + +BENCH_DIR = Path(__file__).resolve().parent +REPO_ROOT = BENCH_DIR.parents[1] +SHIM_DIR = BENCH_DIR / "_shim" +SIMULATOR = BENCH_DIR / "provider_simulator.py" +PROVIDER_NAME = "bench_simulator" +COLUMN_ID = "email" +SCHEMA_VERSION = 1 + + +# ── small helpers ───────────────────────────────────────────────────────── + +def percentile(values: list[float], pct: float) -> float: + """Nearest-rank percentile (same definition as the queue load gate).""" + if not values: + return 0.0 + ordered = sorted(values) + index = min(len(ordered) - 1, max(0, math.ceil(pct * len(ordered)) - 1)) + return ordered[index] + + +def _ms_summary(seconds: list[float]) -> dict: + return { + "count": len(seconds), + "p50": round(percentile(seconds, 0.50) * 1000, 1), + "p95": round(percentile(seconds, 0.95) * 1000, 1), + "p99": round(percentile(seconds, 0.99) * 1000, 1), + "max": round(max(seconds) * 1000, 1) if seconds else 0.0, + } + + +def _run(cmd: list[str]) -> str: + try: + return subprocess.run(cmd, capture_output=True, text=True, timeout=10, check=False, + cwd=REPO_ROOT).stdout.strip() + except Exception: + return "" + + +def _hardware() -> dict: + system = platform.system() + cpu_model, ram_bytes, physical = "", 0, None + if system == "Darwin": + cpu_model = _run(["sysctl", "-n", "machdep.cpu.brand_string"]) + ram_bytes = int(_run(["sysctl", "-n", "hw.memsize"]) or 0) + physical = int(_run(["sysctl", "-n", "hw.physicalcpu"]) or 0) or None + elif system == "Linux": + try: + for line in Path("/proc/cpuinfo").read_text().splitlines(): + if line.startswith("model name"): + cpu_model = line.split(":", 1)[1].strip() + break + for line in Path("/proc/meminfo").read_text().splitlines(): + if line.startswith("MemTotal"): + ram_bytes = int(line.split()[1]) * 1024 + break + except OSError: + pass + return { + "cpu_model": cpu_model or platform.processor() or "unknown", + "physical_cores": physical, + "logical_cores": os.cpu_count(), + "ram_gb": round(ram_bytes / 2**30, 1) if ram_bytes else None, + "os": platform.platform(), + "machine": platform.machine(), + } + + +def _git() -> dict: + return { + "commit": _run(["git", "rev-parse", "HEAD"]) or os.getenv("OPENGTM_BUILD_SHA", ""), + "branch": _run(["git", "rev-parse", "--abbrev-ref", "HEAD"]), + "dirty": bool(_run(["git", "status", "--porcelain", "--untracked-files=no"])), + } + + +def _redis_reachable(url: str) -> bool: + try: + import redis + + return bool(redis.Redis.from_url(url, socket_connect_timeout=0.5).ping()) + except Exception: + return False + + +def _http_json(url: str) -> dict: + with urllib.request.urlopen(url, timeout=5) as response: + return json.loads(response.read()) + + +# ── simulator lifecycle ─────────────────────────────────────────────────── + +def start_simulator(args: argparse.Namespace) -> tuple[subprocess.Popen, str]: + cmd = [ + sys.executable, str(SIMULATOR), "--host", "127.0.0.1", "--port", "0", + "--seed", str(args.seed), "--latency-ms", str(args.latency_ms), + "--latency-dist", args.latency_dist, "--jitter-ms", str(args.jitter_ms), + "--sigma", str(args.sigma), "--error-rate", str(args.error_rate), + "--rate-429", str(args.rate_429), + ] + env = {k: v for k, v in os.environ.items() if not k.startswith("OPENGTM_BENCH_")} + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, env=env) + deadline = time.monotonic() + 20 + assert proc.stdout is not None + while time.monotonic() < deadline: + line = proc.stdout.readline() + if line.startswith("READY "): + return proc, line.split(" ", 1)[1].strip() + if not line and proc.poll() is not None: + break + proc.kill() + raise RuntimeError("provider simulator failed to start") + + +def write_manifest(path: Path, origin: str, args: argparse.Namespace) -> None: + """A normal declarative manifest; only its URL points at the simulator.""" + path.write_text( + f"""manifest_version: "1" +name: {PROVIDER_NAME} +display_name: Benchmark provider simulator +author: OpenGTM bench +capability: email +default_confidence: 0.9 +cost_per_lookup: {args.cost_per_lookup} +auth: + type: none +request: + method: POST + url: {origin}/v1/enrich + headers: + Content-Type: application/json + body: + domain: "{{{{input.domain}}}}" + company: "{{{{input.company}}}}" + timeout: {args.provider_http_timeout} +response: + mappings: + email: "$.data.email" +""", + encoding="utf-8", + ) + + +# ── the measured run ────────────────────────────────────────────────────── + +def _columns_config() -> list[dict]: + return [ + {"id": "company", "name": "Company", "type": "lead_field", "lead_field": "company"}, + {"id": "website", "name": "Website", "type": "lead_field", "lead_field": "website"}, + { + "id": COLUMN_ID, "name": "Email", "type": "enrichment", + "provider": PROVIDER_NAME, "target_field": "email", + # Keep the run to one provider call per cell: the email verify + # cascade does live DNS/SMTP and would measure the network instead. + "verify": False, + }, + ] + + +def _payload(args, workbook_id: str, workspace_id: str) -> dict: + """Mirror POST /workbooks/{id}/run's run_workbook payload.""" + return { + "workbook_id": workbook_id, + "workspace_id": workspace_id, + "run_id": str(uuid.uuid4()), + "column_ids": [COLUMN_ID], + "row_columns": None, + "row_ids": None, + "lead_ids": None, + "concurrency": args.row_concurrency, + "max_providers": args.max_providers, + "retry_passes": args.retry_passes, + "provider_workers": args.provider_workers, + "provider_timeout": args.provider_timeout, + "fill_missing": False, + "force": False, + } + + +async def run_benchmark(args: argparse.Namespace, timings_path: Path) -> dict: + from apps.api.database import SessionLocal, engine + from apps.api.db_init import init_db + from apps.api.models import Job + from apps.api.services.job_registry import register_job_handlers + from apps.api.services.queue_service import JOB_TIMEOUTS, QueueService + from apps.api.services.workbook.models import ( + Workbook, + WorkbookEnrichment, + WorkbookRow, + ) + + init_db() + tag = f"bench-enrich:{uuid.uuid4().hex[:12]}:" + with SessionLocal() as db: + if not args.allow_busy_queue and db.query(Job).filter( + Job.status.in_(["pending", "processing"]) + ).count(): + raise RuntimeError("queue has pending/processing jobs; use a disposable database") + + workbook_ids: list[str] = [] + try: + # ── fixtures: deterministic rows, stable simulator keys ── + with SessionLocal() as db: + for w in range(args.workbooks): + workbook_id = f"{tag}wb{w}".replace(":", "-") + workbook_ids.append(workbook_id) + db.add(Workbook( + id=workbook_id, workspace_id=f"bench-ws-{w % args.workspaces}", + name=f"Enrichment bench {w}", source_type="empty", + columns_config=_columns_config(), status="draft", + )) + db.flush() + rows = [] + for i in range(args.rows): + w = i % args.workbooks + rows.append({ + "workspace_id": f"bench-ws-{w % args.workspaces}", + "workbook_id": workbook_ids[w], "position": i // args.workbooks, + "data": {"company": f"Bench Co {i:07d}", + "website": f"https://bench-{i:07d}.example"}, + "enrichments": {}, + }) + db.bulk_insert_mappings(WorkbookRow, rows) + db.commit() + + # ── queue + worker (the production QueueService, scoped to our jobs) ── + queue = QueueService(concurrency=args.worker_concurrency) + queue.max_active_per_workspace = args.max_active_per_workspace + register_job_handlers(queue) + queue.claim_next_job = functools.partial(queue.claim_next_job, fire_key_prefix=tag) + + enqueued_at = time.time() + with SessionLocal() as db: + for w, workbook_id in enumerate(workbook_ids): + wb = db.query(Workbook).filter(Workbook.id == workbook_id).one() + wb.status, wb.completed_rows = "running", 0 + wb.total_rows = sum(1 for i in range(args.rows) if i % args.workbooks == w) + db.commit() + queue.add_job(db, "run_workbook", _payload(args, workbook_id, wb.workspace_id), + fire_key=f"{tag}{w}") + + await queue.start_worker() + deadline = time.monotonic() + args.timeout + timed_out = False + try: + while True: + with SessionLocal() as db: + active = db.query(Job).filter( + Job.fire_key.like(f"{tag}%"), + Job.status.in_(["pending", "processing"]), + ).count() + if not active: + break + if time.monotonic() > deadline: + timed_out = True + break + await asyncio.sleep(0.2) + finally: + await queue.stop_worker() + finished_at = time.time() + + # ── collect results ── + with SessionLocal() as db: + jobs = db.query(Job).filter(Job.fire_key.like(f"{tag}%")).all() + job_statuses = Counter(job.status for job in jobs) + job_errors = [job.error for job in jobs if job.error][:5] + job_attempts = sum((job.retry_count or 0) for job in jobs) + cell_status: Counter[str] = Counter() + cell_errors: Counter[str] = Counter() + for (enrichments,) in db.query(WorkbookRow.enrichments).filter( + WorkbookRow.workbook_id.in_(workbook_ids) + ): + cell = (enrichments or {}).get(COLUMN_ID) + status = cell.get("status") if isinstance(cell, dict) else "missing" + cell_status[status] += 1 + if status != "complete": + cell_errors[str((cell or {}).get("error") or status)[:80]] += 1 + overlay_rows = db.query(WorkbookEnrichment).filter( + WorkbookEnrichment.workbook_id.in_(workbook_ids) + ).count() + dialect = engine.dialect.name + if dialect == "sqlite": + import sqlite3 + + db_version = sqlite3.sqlite_version + else: + from sqlalchemy import text + + db_version = db.execute(text("SHOW server_version")).scalar() + + records = [] + if timings_path.exists(): + records = [json.loads(line) for line in timings_path.read_text().splitlines() if line] + final_by_cell: dict[tuple, dict] = {} + for record in sorted(records, key=lambda r: r["end"]): + final_by_cell[(record["workbook_id"], record["row_id"])] = record + finals = list(final_by_cell.values()) + completed = cell_status.get("complete", 0) + last_end = max((r["end"] for r in finals), default=finished_at) + first_start = min((r["start"] for r in records), default=enqueued_at) + busy = max(1e-9, last_end - first_start) + to_last = max(1e-9, last_end - enqueued_at) + + return { + "timed_out": timed_out, + "dialect": dialect, + "db_version": db_version, + "jobs": {"count": len(jobs), "statuses": dict(job_statuses), + "retries": job_attempts, "errors": job_errors}, + "cells": { + "total": args.rows, "completed": completed, + "statuses": dict(cell_status), "errors": dict(cell_errors.most_common(10)), + "overlay_rows": overlay_rows, "attempts_recorded": len(records), + }, + "throughput": { + "completed_per_second": round(completed / to_last, 2), + "completed_per_second_excluding_startup": round(completed / busy, 2), + }, + "latency_ms": { + "end_to_end": _ms_summary([r["end"] - enqueued_at for r in finals]), + "cell_execution": _ms_summary([r["end"] - r["start"] for r in records]), + "startup_to_first_cell": round((first_start - enqueued_at) * 1000, 1), + }, + "wall_seconds": { + "enqueue_to_last_cell": round(to_last, 3), + "enqueue_to_jobs_done": round(finished_at - enqueued_at, 3), + }, + "job_timeout_seconds": JOB_TIMEOUTS.get("run_workbook"), + } + finally: + if not args.keep_data and workbook_ids: + with SessionLocal() as db: + db.query(WorkbookEnrichment).filter( + WorkbookEnrichment.workbook_id.in_(workbook_ids)).delete(synchronize_session=False) + db.query(WorkbookRow).filter( + WorkbookRow.workbook_id.in_(workbook_ids)).delete(synchronize_session=False) + db.query(Workbook).filter( + Workbook.id.in_(workbook_ids)).delete(synchronize_session=False) + db.query(Job).filter(Job.fire_key.like(f"{tag}%")).delete(synchronize_session=False) + db.commit() + + +def _resources() -> dict: + own = resource.getrusage(resource.RUSAGE_SELF) + kids = resource.getrusage(resource.RUSAGE_CHILDREN) + # ru_maxrss is bytes on macOS, KiB on Linux. + scale = 1 if platform.system() == "Darwin" else 1024 + return { + "harness_cpu_seconds": round(own.ru_utime + own.ru_stime, 2), + "children_cpu_seconds": round(kids.ru_utime + kids.ru_stime, 2), + "harness_peak_rss_mb": round(own.ru_maxrss * scale / 2**20, 1), + "largest_child_peak_rss_mb": round(kids.ru_maxrss * scale / 2**20, 1), + "note": "children = job child + provider workers + simulator (reaped processes)", + } + + +def summary_text(report: dict) -> str: + r, w, s = report["results"], report["identity"]["workload"], report["identity"]["worker_config"] + lat = r["latency_ms"] + sim = report.get("simulator_stats") or {} + lines = [ + f"OpenGTM enrichment baseline ok={report['ok']} commit={report['identity']['git']['commit'][:12]}", + f" workload : {w['rows']} rows x 1 declarative HTTP column, {w['workbooks']} workbook(s), " + f"simulator {w['simulator']['latency_dist']} {w['simulator']['latency_ms']}ms " + f"err={w['simulator']['error_rate']} 429={w['simulator']['rate_429']} seed={w['simulator']['seed']}", + f" workers : slots={s['worker_concurrency']} ws_cap={s['max_active_per_workspace']} " + f"rows||={s['row_concurrency']} provider_workers={s['provider_workers']} " + f"retry_passes={s['retry_passes']} db={r['dialect']}", + f" result : {r['cells']['completed']}/{r['cells']['total']} completed, " + f"{r['throughput']['completed_per_second']} completions/s " + f"({r['throughput']['completed_per_second_excluding_startup']}/s excl. startup)", + f" latency : e2e p50/p95/p99 = {lat['end_to_end']['p50']}/{lat['end_to_end']['p95']}/" + f"{lat['end_to_end']['p99']} ms; cell p50/p95/p99 = {lat['cell_execution']['p50']}/" + f"{lat['cell_execution']['p95']}/{lat['cell_execution']['p99']} ms; " + f"startup {lat['startup_to_first_cell']} ms", + f" errors : {r['cells']['errors'] or 'none'}; provider requests={sim.get('requests')} " + f"statuses={sim.get('status_counts')}", + f" wall : {r['wall_seconds']['enqueue_to_jobs_done']} s", + ] + return "\n".join(lines) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + p = argparse.ArgumentParser(description="Enrichment completion harness (M0 baseline)") + p.add_argument("--rows", type=int, default=500) + p.add_argument("--workbooks", type=int, default=1, help="rows are split round-robin; one run_workbook job each") + p.add_argument("--workspaces", type=int, default=1, help="workbooks are assigned round-robin") + # simulator + p.add_argument("--seed", type=int, default=42) + p.add_argument("--latency-ms", type=float, default=50.0) + p.add_argument("--latency-dist", choices=("fixed", "uniform", "lognormal"), default="fixed") + p.add_argument("--jitter-ms", type=float, default=0.0) + p.add_argument("--sigma", type=float, default=0.25) + p.add_argument("--error-rate", type=float, default=0.0) + p.add_argument("--rate-429", type=float, default=0.0) + p.add_argument("--cost-per-lookup", type=float, default=0.0, + help=">0 routes calls through the spend-reservation path") + p.add_argument("--provider-http-timeout", type=float, default=20.0) + # worker config (defaults = shipped defaults; see docs/operations/worker-concurrency.md) + p.add_argument("--worker-concurrency", type=int, default=None, help="default: WORKER_CONCURRENCY") + p.add_argument("--max-active-per-workspace", type=int, default=None, + help="default: WORKER_MAX_ACTIVE_PER_WORKSPACE") + p.add_argument("--row-concurrency", type=int, default=None, help="default: Settings row_concurrency") + p.add_argument("--provider-workers", type=int, default=None, help="default: Settings provider_workers") + p.add_argument("--provider-timeout", type=float, default=None, help="default: Settings provider_timeout") + p.add_argument("--max-providers", type=int, default=None, help="default: Settings max_providers") + p.add_argument("--retry-passes", type=int, default=None, help="default: Settings retry_passes") + # environment + p.add_argument("--database-url", default=os.getenv("BENCH_DATABASE_URL", ""), + help="disposable DB; default is a temporary SQLite file") + p.add_argument("--allow-busy-queue", action="store_true") + p.add_argument("--timeout", type=float, default=1800.0, help="overall wait for jobs (s)") + p.add_argument("--keep-data", action="store_true", help="keep DB rows and temp dir") + p.add_argument("--output", help="write the JSON report here") + p.add_argument("--json", action="store_true", help="print the JSON report to stdout") + p.add_argument("--label", default="", help="free-form label stored in the report") + args = p.parse_args(argv) + if not 1 <= args.rows <= 200_000 or not 1 <= args.workbooks <= min(args.rows, 64): + p.error("rows must be 1..200000 and workbooks 1..min(rows, 64)") + if not 1 <= args.workspaces <= args.workbooks: + p.error("workspaces must be 1..workbooks") + return args + + +def _resolve_worker_defaults(args: argparse.Namespace) -> None: + """Fill unset knobs from the same sources production uses.""" + from apps.api.core.config import settings + from apps.api.routers.settings import _ENRICHMENT_DEFAULTS + + if args.worker_concurrency is None: + args.worker_concurrency = settings.WORKER_CONCURRENCY + if args.max_active_per_workspace is None: + args.max_active_per_workspace = settings.WORKER_MAX_ACTIVE_PER_WORKSPACE + for key in ("row_concurrency", "provider_workers", "provider_timeout", + "max_providers", "retry_passes"): + if getattr(args, key) is None: + setattr(args, key, _ENRICHMENT_DEFAULTS[key]) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + workdir = Path(tempfile.mkdtemp(prefix="opengtm-bench-")) + database_url = args.database_url or f"sqlite:///{workdir}/bench.db" + os.environ["DATABASE_URL"] = database_url + os.environ.setdefault("LOG_LEVEL", "WARNING") + sys.path.insert(0, str(REPO_ROOT)) + + simulator, origin = start_simulator(args) + manifest_path = workdir / f"{PROVIDER_NAME}.yaml" + timings_path = workdir / "cell_timings.jsonl" + write_manifest(manifest_path, origin, args) + os.environ.update({ + "OPENGTM_BENCH_SIMULATOR_ORIGIN": origin, + "OPENGTM_BENCH_MANIFEST": str(manifest_path), + "OPENGTM_BENCH_TIMINGS": str(timings_path), + "PYTHONPATH": os.pathsep.join( + [str(SHIM_DIR), str(REPO_ROOT)] + [p for p in [os.getenv("PYTHONPATH", "")] if p]), + }) + + started = time.time() + try: + _resolve_worker_defaults(args) + results = asyncio.run(run_benchmark(args, timings_path)) + try: + simulator_stats = _http_json(f"{origin}/stats") + except Exception as exc: # pragma: no cover - diagnostic only + simulator_stats = {"error": str(exc)} + finally: + simulator.terminate() + try: + simulator.wait(timeout=5) + except subprocess.TimeoutExpired: + simulator.kill() + if not args.keep_data: + shutil.rmtree(workdir, ignore_errors=True) + + from apps.api.services.queue_service import QueueService + + redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") + report = { + "schema_version": SCHEMA_VERSION, + "benchmark": "enrichment_completion", + "label": args.label, + "ok": not results["timed_out"] and results["cells"]["completed"] > 0, + "started_at": datetime.fromtimestamp(started, UTC).isoformat(), + "identity": { + "git": _git(), + "hardware": _hardware(), + "python": {"version": platform.python_version(), + "implementation": platform.python_implementation()}, + "database": {"dialect": results["dialect"], "version": results["db_version"], + "location": "temporary file" if not args.database_url else "caller-supplied"}, + "redis": {"reachable": _redis_reachable(redis_url)}, + "worker_config": { + "worker_concurrency": args.worker_concurrency, + "max_active_per_workspace": args.max_active_per_workspace, + "worker_processes": 1, + "heartbeat_interval_seconds": QueueService(concurrency=1).heartbeat_interval, + "run_workbook_timeout_seconds": results["job_timeout_seconds"], + "row_concurrency": args.row_concurrency, + "provider_workers": args.provider_workers, + "provider_timeout_seconds": args.provider_timeout, + "max_providers": args.max_providers, + "retry_passes": args.retry_passes, + }, + "workload": { + "rows": args.rows, "workbooks": args.workbooks, "workspaces": args.workspaces, + "columns": [{"type": "enrichment", "provider": PROVIDER_NAME, + "path": "declarative manifest -> pebble provider worker -> httpx"}], + "cost_per_lookup": args.cost_per_lookup, + "simulator": {"seed": args.seed, "latency_ms": args.latency_ms, + "latency_dist": args.latency_dist, "jitter_ms": args.jitter_ms, + "sigma": args.sigma, "error_rate": args.error_rate, + "rate_429": args.rate_429}, + }, + }, + "results": {k: v for k, v in results.items() if k not in ("dialect", "db_version", "job_timeout_seconds")}, + "simulator_stats": {k: v for k, v in simulator_stats.items() if k != "config"}, + "resources": _resources(), + } + report["results"]["dialect"] = results["dialect"] + + encoded = json.dumps(report, indent=2, sort_keys=False) + if args.output: + out = Path(args.output) + out.parent.mkdir(parents=True, exist_ok=True) + tmp = out.with_suffix(out.suffix + ".tmp") + tmp.write_text(encoded + "\n", encoding="utf-8") + tmp.replace(out) + if args.json: + print(encoded) + print(summary_text(report), file=sys.stderr if args.json else sys.stdout) + return 0 if report["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/bench/provider_simulator.py b/scripts/bench/provider_simulator.py new file mode 100755 index 00000000..33bfb05e --- /dev/null +++ b/scripts/bench/provider_simulator.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""Deterministic enrichment-provider simulator for benchmarks. + +A tiny local HTTP server that behaves like a paid enrichment API with a +configurable, *seeded* latency distribution, 5xx error rate, and 429 rate. It +returns fixed enrichment payloads so throughput runs never touch a real vendor. + +Determinism: every response is a pure function of ``(seed, request key, +attempt number for that key)``. The key is the request's ``domain`` field, so +outcomes do not depend on the order in which concurrent requests arrive. A retry +of the same key draws a fresh (but still reproducible) outcome. + + uv run python scripts/bench/provider_simulator.py --port 8099 \ + --latency-ms 50 --latency-dist lognormal --sigma 0.25 \ + --error-rate 0.01 --rate-429 0.02 --seed 42 + +Endpoints: + POST /v1/enrich {"domain": "...", "company": "..."} -> {"data": {...}} + GET /stats request/status counters (for harness reports) + GET /healthz liveness + +On start it prints ``READY http://HOST:PORT`` (one line) to stdout so a parent +process can discover an ephemeral port (``--port 0``). + +Benchmark-only tooling: this is not part of the application package. +""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import math +import random +import sys +import time +from collections import Counter +from dataclasses import asdict, dataclass + +from aiohttp import web + +LATENCY_DISTRIBUTIONS = ("fixed", "uniform", "lognormal") + + +@dataclass(frozen=True) +class SimulatorConfig: + seed: int = 42 + latency_ms: float = 0.0 + latency_dist: str = "fixed" + # uniform: +/- jitter_ms around latency_ms. lognormal: sigma of ln(latency). + jitter_ms: float = 0.0 + sigma: float = 0.25 + error_rate: float = 0.0 + rate_429: float = 0.0 + + def validate(self) -> SimulatorConfig: + if self.latency_dist not in LATENCY_DISTRIBUTIONS: + raise ValueError(f"latency_dist must be one of {LATENCY_DISTRIBUTIONS}") + if self.latency_ms < 0 or self.jitter_ms < 0 or self.sigma < 0: + raise ValueError("latency parameters must be non-negative") + if not (0 <= self.error_rate <= 1 and 0 <= self.rate_429 <= 1): + raise ValueError("error_rate and rate_429 must be within [0, 1]") + if self.error_rate + self.rate_429 > 1: + raise ValueError("error_rate + rate_429 must not exceed 1") + return self + + +def decide(config: SimulatorConfig, key: str, attempt: int) -> tuple[float, int]: + """Return ``(latency_seconds, http_status)`` for one request. + + Pure and order-independent: the RNG is seeded from a stable digest of the + seed, the request key, and the per-key attempt number. + """ + digest = hashlib.sha256(f"{config.seed}:{key}:{attempt}".encode()).digest() + rng = random.Random(int.from_bytes(digest[:8], "big")) + base = config.latency_ms + if config.latency_dist == "uniform": + latency_ms = rng.uniform(base - config.jitter_ms, base + config.jitter_ms) + elif config.latency_dist == "lognormal" and base > 0: + latency_ms = rng.lognormvariate(math.log(base), config.sigma) + else: + latency_ms = base + roll = rng.random() + if roll < config.rate_429: + status = 429 + elif roll < config.rate_429 + config.error_rate: + status = 500 + else: + status = 200 + return max(0.0, latency_ms) / 1000.0, status + + +def payload_for(domain: str, company: str) -> dict: + """The fixed enrichment payload returned on success.""" + slug = domain or "unknown.invalid" + return { + "data": { + "email": f"contact@{slug}", + "company": company or slug, + "company_size": "51-200", + "industry": "Software", + } + } + + +def build_app(config: SimulatorConfig) -> web.Application: + config.validate() + attempts: Counter[str] = Counter() + statuses: Counter[int] = Counter() + state = {"requests": 0, "started_at": time.time(), "in_flight": 0, "peak_in_flight": 0} + + async def enrich(request: web.Request) -> web.Response: + try: + body = await request.json() + except Exception: + body = {} + domain = str((body or {}).get("domain") or request.query.get("domain") or "") + company = str((body or {}).get("company") or "") + attempt = attempts[domain] + attempts[domain] += 1 + state["requests"] += 1 + state["in_flight"] += 1 + state["peak_in_flight"] = max(state["peak_in_flight"], state["in_flight"]) + try: + latency, status = decide(config, domain, attempt) + if latency: + await asyncio.sleep(latency) + finally: + state["in_flight"] -= 1 + statuses[status] += 1 + if status == 429: + return web.json_response({"error": "rate_limited"}, status=429, + headers={"Retry-After": "1"}) + if status >= 500: + return web.json_response({"error": "simulated_failure"}, status=status) + return web.json_response(payload_for(domain, company)) + + async def stats(_request: web.Request) -> web.Response: + return web.json_response({ + "config": asdict(config), + "requests": state["requests"], + "unique_keys": len(attempts), + "retried_keys": sum(1 for count in attempts.values() if count > 1), + "status_counts": {str(code): count for code, count in sorted(statuses.items())}, + "peak_in_flight": state["peak_in_flight"], + "uptime_seconds": round(time.time() - state["started_at"], 3), + }) + + async def healthz(_request: web.Request) -> web.Response: + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_post("/v1/enrich", enrich) + app.router.add_get("/v1/enrich", enrich) + app.router.add_get("/stats", stats) + app.router.add_get("/healthz", healthz) + return app + + +async def _serve(config: SimulatorConfig, host: str, port: int) -> None: + runner = web.AppRunner(build_app(config), access_log=None) + await runner.setup() + # Large backlog: benchmark clients open many short-lived connections. + site = web.TCPSite(runner, host, port, backlog=4096) + await site.start() + bound = site._server.sockets[0].getsockname()[1] # type: ignore[union-attr] + print(f"READY http://{host}:{bound}", flush=True) + try: + await asyncio.Event().wait() + finally: + await runner.cleanup() + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=0, help="0 = ephemeral") + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--latency-ms", type=float, default=0.0) + parser.add_argument("--latency-dist", choices=LATENCY_DISTRIBUTIONS, default="fixed") + parser.add_argument("--jitter-ms", type=float, default=0.0) + parser.add_argument("--sigma", type=float, default=0.25) + parser.add_argument("--error-rate", type=float, default=0.0) + parser.add_argument("--rate-429", type=float, default=0.0) + return parser.parse_args(argv) + + +def config_from_args(args: argparse.Namespace) -> SimulatorConfig: + return SimulatorConfig( + seed=args.seed, latency_ms=args.latency_ms, latency_dist=args.latency_dist, + jitter_ms=args.jitter_ms, sigma=args.sigma, error_rate=args.error_rate, + rate_429=args.rate_429, + ).validate() + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + try: + asyncio.run(_serve(config_from_args(args), args.host, args.port)) + except KeyboardInterrupt: + pass + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_enrichment_bench_harness.py b/tests/test_enrichment_bench_harness.py new file mode 100644 index 00000000..401b977f --- /dev/null +++ b/tests/test_enrichment_bench_harness.py @@ -0,0 +1,74 @@ +"""Keep the M0 enrichment benchmark harness and provider simulator working.""" + +from __future__ import annotations + +import importlib.util +import json +import os +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +BENCH_DIR = REPO_ROOT / "scripts" / "bench" + + +def _load(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[name] = module # dataclasses resolve their module by name + spec.loader.exec_module(module) + return module + + +def test_simulator_outcomes_are_deterministic_and_order_independent(): + sim = _load("bench_provider_simulator", BENCH_DIR / "provider_simulator.py") + config = sim.SimulatorConfig(seed=7, latency_ms=50, latency_dist="lognormal", + sigma=0.5, error_rate=0.1, rate_429=0.2).validate() + keys = [f"bench-{i:07d}.example" for i in range(2000)] + forward = [sim.decide(config, key, 0) for key in keys] + backward = [sim.decide(config, key, 0) for key in reversed(keys)][::-1] + assert forward == backward + statuses = [status for _, status in forward] + assert 0.15 < statuses.count(429) / len(keys) < 0.25 + assert 0.06 < statuses.count(500) / len(keys) < 0.14 + # A retry of the same key draws a fresh, reproducible outcome. + assert sim.decide(config, keys[0], 1) == sim.decide(config, keys[0], 1) + assert sim.decide(sim.SimulatorConfig(latency_ms=50), "k", 0) == (0.05, 200) + + +def test_shim_only_exempts_loopback_origins(): + shim = _load("bench_sitecustomize", BENCH_DIR / "_shim" / "sitecustomize.py") + assert shim._is_loopback_origin("http://127.0.0.1:8099") + assert not shim._is_loopback_origin("http://10.0.0.5:8099") + assert not shim._is_loopback_origin("http://169.254.169.254:80") + assert not shim._is_loopback_origin("http://localhost:8099") + assert not shim._is_loopback_origin("https://127.0.0.1") + + +def test_harness_smoke_completes_tiny_run(tmp_path): + """End to end: queue claim -> job child -> declarative provider -> commit.""" + output = tmp_path / "report.json" + env = {k: v for k, v in os.environ.items() + if k not in {"DATABASE_URL", "BENCH_DATABASE_URL", "PYTHONPATH"} + and not k.startswith("OPENGTM_BENCH_")} + result = subprocess.run( + [sys.executable, str(BENCH_DIR / "enrichment_harness.py"), "--rows", "5", + "--latency-ms", "0", "--timeout", "60", "--output", str(output)], + cwd=REPO_ROOT, env=env, capture_output=True, text=True, timeout=120, check=False, + ) + assert result.returncode == 0, result.stderr[-4000:] + report = json.loads(output.read_text()) + assert report["ok"] is True + assert report["results"]["cells"]["completed"] == 5 + assert report["results"]["jobs"]["statuses"] == {"completed": 1} + assert report["simulator_stats"]["requests"] == 5 + assert report["simulator_stats"]["status_counts"] == {"200": 5} + assert report["results"]["latency_ms"]["end_to_end"]["count"] == 5 + identity = report["identity"] + assert identity["database"]["dialect"] == "sqlite" + assert identity["workload"]["rows"] == 5 + assert {"worker_concurrency", "max_active_per_workspace", "row_concurrency", + "provider_workers"} <= set(identity["worker_config"]) + assert identity["hardware"]["logical_cores"]