Add SDK-backed performance scenarios and C++ performance driver

.github/workflows/ci.yml

@@ -106,13 +106,31 @@ jobs:
   performance-tests:
     name: performance tests
     runs-on: ubuntu-24.04
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: a2a
+          POSTGRES_PASSWORD: a2a
+          POSTGRES_DB: a2a
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U a2a -d a2a"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      A2A_TEST_POSTGRES_DSN: postgresql://a2a:a2a@127.0.0.1:5432/a2a
     steps:
       - uses: actions/checkout@v6
-      - name: Run report-only performance smoke suite
+      - name: Install deps
+        run: ./scripts/install_build_deps.sh
+      - name: Run report-only performance suite
         env:
           A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json
           A2A_PERF_STORE_BACKENDS: inmemory,postgres
-          A2A_PERF_REQUESTS: 100000
+          A2A_PERF_REQUESTS: 1000
           A2A_PERF_CONCURRENCY: 1,4
           A2A_PERF_WARMUP_SECONDS: 0
           A2A_PERF_REPORT_DIR: perf-artifacts

docs/performance-testing.md

@@ -1,15 +1,14 @@
 # Performance testing
 
 The A2A C++ SDK includes a report-only performance test kit for repeatable local
-and CI measurements across TCK-aligned SDK operation paths. The first version is
-intended to produce consistent artifacts without introducing threshold gates;
-future revisions can replace individual in-process drivers with deeper
-wire-level probes and baseline comparisons.
+and CI measurements. The Python runner still owns matrix orchestration and writes
+`results.json`, `results.csv`, and `summary.md`; measured operations are delegated
+to the C++ SDK-backed driver built as `a2a_performance_driver`.
 
 ## Run locally
 
 ```bash
-./scripts/run_performance_tests.sh
+./scripts/run_performance_tests.sh --store-backends inmemory --requests 100 --concurrency 1,4 --warmup-seconds 0
 ```
 
 The runner writes:
@@ -18,6 +17,9 @@ The runner writes:
 - `perf-artifacts/results.csv`
 - `perf-artifacts/summary.md`
 
+If `A2A_PERF_DRIVER` is not set, the runner configures and builds the driver in
+`build/performance`. Set `A2A_PERF_BUILD_DIR` to reuse another CMake build tree.
+
 ## Configuration
 
 You can configure the matrix and load profile with command-line flags or the
@@ -32,56 +34,70 @@ matching environment variables.
 | Warmup seconds | `A2A_PERF_WARMUP_SECONDS` | `1` |
 | Duration seconds metadata | `A2A_PERF_DURATION_SECONDS` | `0` |
 | Report directory | `A2A_PERF_REPORT_DIR` | `perf-artifacts` |
+| Existing driver binary | `A2A_PERF_DRIVER` | unset |
+| Auto-build directory | `A2A_PERF_BUILD_DIR` | `build/performance` |
+
+## Scenario coverage
+
+The SDK-backed driver executes these real service/store flows through the shared
+example executor, task lifecycle service, task subscription service, in-memory
+task store, in-memory push notification store, and push notification service:
+
+- task lifecycle: `SendMessage` create, `GetTask`, `CancelTask`, list with and
+  without pagination, follow-up `SendMessage`, and missing-task lookup;
+- streaming and subscriptions: finite `SendStreamingMessage`, existing-task
+  subscription first event, multi-subscriber reads, terminal stream completion,
+  and a disconnect-style subscriber read where other subscriptions still run;
+- push notifications: create/get/list/delete config, notify many configs for one
+  task update, and delivery callback latency through a local recording delivery
+  client implementation of the SDK delivery interface.
+
+Each row includes the required stable fields plus `driver_type` and
+`transport_path`. The current driver type is `cpp_sdk_in_process`.
+
+## Transport coverage
 
-Example smoke run:
+The current C++ driver is SDK-backed and in-process. It validates the selected
+transport and records a transport-specific `transport_path` (`sdk_grpc_server_dispatch`,
+`sdk_jsonrpc_server_dispatch`, or `sdk_http_json_server_dispatch`) so reports stay
+compatible with the existing transport matrix. It does not yet open sockets or
+run external clients; k6 is not required. Future work can replace individual
+transport paths with wire-level gRPC, JSON-RPC, and HTTP+JSON probes without
+changing the report format.
+
+## Store backend coverage
+
+`inmemory` uses the real in-memory task and push notification stores. `postgres`
+uses the repository `PostgresStoreFactory` to create real PostgreSQL-backed task
+and push notification stores when the driver is built with
+`A2A_ENABLE_POSTGRES_STORE=ON`. The Python runner automatically adds that CMake
+option when `postgres` is selected and it needs to auto-build the driver.
+PostgreSQL runs must provide the same local DSN style used by the repository
+store tests (`A2A_TEST_POSTGRES_DSN`); CI starts a local PostgreSQL service for
+the performance job.
+
+## CI behavior
+
+The performance job remains report-only: it uploads `perf-artifacts`, appends
+`summary.md` to the GitHub Actions step summary, and fails only on crashes,
+functional operation errors, malformed output, or missing artifacts. It does not
+enforce latency or throughput thresholds.
+
+## Larger local benchmark
 
 ```bash
-A2A_PERF_TRANSPORTS=grpc \
-A2A_PERF_STORE_BACKENDS=inmemory \
-A2A_PERF_REQUESTS=100 \
-A2A_PERF_CONCURRENCY=1,4 \
-A2A_PERF_WARMUP_SECONDS=0 \
+A2A_PERF_TRANSPORTS=grpc,jsonrpc,http_json \
+A2A_PERF_STORE_BACKENDS=inmemory,postgres \
+A2A_PERF_REQUESTS=1000 \
+A2A_PERF_CONCURRENCY=1,4,16,64 \
+A2A_PERF_WARMUP_SECONDS=5 \
+A2A_PERF_REPORT_DIR=perf-artifacts \
 ./scripts/run_performance_tests.sh
 ```
 
-## Scenario coverage
-
-The suite emits one result object for each scenario, transport, store backend,
-and concurrency combination. Scenarios are aligned with TCK-relevant behavior:
-
-- task lifecycle: send, get, cancel, list, follow-up send, and missing-task error;
-- streaming and subscriptions: finite streaming send, first-event latency,
-  multiple subscribers, terminal completion, and subscriber disconnect;
-- push notifications: configuration create/get/list/delete, many-config update
-  notification, and fake local callback delivery latency;
-- transport labels: gRPC, JSON-RPC, and HTTP+JSON;
-- store labels: in-memory and PostgreSQL-backed task/push notification stores.
-
-## Report-only CI behavior
-
-The CI performance job runs a larger report-only matrix with 100 000 operations per
-result row, appends `summary.md` to the GitHub Actions step summary, and uploads
-the `perf-artifacts` directory. The summary starts with a scenario rollup table
-so reviewers can quickly compare
-operations, success/error counts, throughput, and worst observed latency before
-opening the full detailed matrix results. It fails only if the runner crashes,
-reports operation errors, or fails to produce valid report files. It does not
-compare against latency or throughput thresholds yet.
-
-## k6 consideration
-
-k6 is a strong candidate for future wire-level HTTP+JSON and JSON-RPC load
-testing because it has mature virtual-user orchestration and CI-friendly
-summaries. The first version intentionally keeps the runner in Python so it has
-no new runtime dependency, can cover non-HTTP transport/store labels in one
-matrix, and remains deterministic in local and CI smoke runs. When we add real
-network probes, k6 can be introduced alongside this runner for HTTP-based
-scenarios while gRPC and store-specific scenarios continue to use SDK-native
-drivers.
-
 ## JSON shape
 
 `results.json` contains host metadata and a `results` array. Each result includes
-scenario name, transport, store backend, concurrency, operation counts,
-throughput, warmup-adjusted latency percentiles, max latency, SDK commit SHA in
+scenario name, transport, store backend, concurrency, operation counts, success
+and error counts, throughput, latency percentiles, max latency, SDK commit SHA in
 metadata, and host OS/CPU metadata.

scripts/performance_runner.py

@@ -1,11 +1,8 @@
 #!/usr/bin/env python3
 """Report-only A2A SDK performance test kit runner.
 
-The first kit version executes deterministic in-process scenarios that mirror the
-TCK operation matrix and records machine-readable smoke/performance reports. The
-scenario bodies intentionally avoid external services so local and CI runs are
-repeatable; transport/store labels define the matrix under test while later
-iterations can swap individual scenario drivers for full wire-level clients.
+The runner orchestrates the matrix and report generation while delegating measured
+operations to the C++ SDK-backed performance driver.
 """
 
 from __future__ import annotations
@@ -17,8 +14,6 @@
 import platform
 import subprocess
 import sys
-import time
-from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Iterable
@@ -45,12 +40,13 @@
     "PushNotify_ManyConfigsOneTaskUpdate",
     "PushDelivery_CallbackLatency",
 )
-DEFAULT_REQUESTS = 1_000
+DEFAULT_REQUESTS = 10_000
 DEFAULT_CONCURRENCY = (1, 4)
+DEFAULT_BUILD_DIR = "build/performance"
+DRIVER_NAME = "a2a_performance_driver"
 DEFAULT_WARMUP_SECONDS = 1.0
 DEFAULT_DURATION_SECONDS = 0.0
 DEFAULT_REPORT_DIR = "perf-artifacts"
-NANOSECONDS_PER_MILLISECOND = 1_000_000.0
 
 
 @dataclass(frozen=True)
@@ -64,46 +60,53 @@ class RunnerConfig:
     report_dir: Path
 
 
-class ScenarioState:
-    def __init__(self) -> None:
-        self.tasks: dict[str, str] = {}
-        self.push_configs: dict[str, str] = {}
-
-    def execute(self, scenario: str, operation_index: int) -> bool:
-        task_id = f"task-{operation_index % 257}"
-        config_id = f"config-{operation_index % 31}"
-        if scenario == "SendMessage_CreateTask":
-            self.tasks[task_id] = "working"
-        elif scenario == "GetTask_ExistingTask":
-            self.tasks.setdefault(task_id, "working")
-            _ = self.tasks[task_id]
-        elif scenario == "CancelTask_WorkingTask":
-            self.tasks[task_id] = "canceled"
-        elif scenario.startswith("ListTasks_"):
-            _ = list(self.tasks.items())[:25]
-        elif scenario == "SendMessage_FollowUpExistingTask":
-            self.tasks[task_id] = "updated"
-        elif scenario == "GetTask_MissingTaskError":
-            return f"missing-{operation_index}" not in self.tasks
-        elif scenario.startswith("SubscribeToTask_") or scenario == "SendStreamingMessage_FiniteStream":
-            self.tasks.setdefault(task_id, "working")
-            _ = (self.tasks[task_id], operation_index % 8)
-        elif scenario == "PushConfig_Create":
-            self.push_configs[config_id] = "http://127.0.0.1/callback"
-        elif scenario == "PushConfig_Get":
-            self.push_configs.setdefault(config_id, "http://127.0.0.1/callback")
-            _ = self.push_configs[config_id]
-        elif scenario == "PushConfig_List":
-            _ = list(self.push_configs.values())
-        elif scenario == "PushConfig_Delete":
-            self.push_configs.pop(config_id, None)
-        elif scenario.startswith("Push"):
-            self.push_configs.setdefault(config_id, "http://127.0.0.1/callback")
-            _ = sum(len(value) for value in self.push_configs.values())
-        else:
-            raise ValueError(f"unknown scenario: {scenario}")
-        return True
 
+def driver_path_from_build(build_dir: Path) -> Path:
+    candidates = [build_dir / "tests" / DRIVER_NAME, build_dir / DRIVER_NAME]
+    for candidate in candidates:
+        if candidate.exists():
+            return candidate
+    return candidates[0]
+
+
+def ensure_driver(config: RunnerConfig) -> Path:
+    explicit = os.environ.get("A2A_PERF_DRIVER")
+    if explicit:
+        driver = Path(explicit)
+        if not driver.exists():
+            raise ValueError(f"A2A_PERF_DRIVER does not exist: {driver}")
+        return driver
+    build_dir = Path(os.environ.get("A2A_PERF_BUILD_DIR", DEFAULT_BUILD_DIR))
+    driver = driver_path_from_build(build_dir)
+    if driver.exists():
+        return driver
+    configure = ["cmake", "-S", ".", "-B", str(build_dir), "-DCMAKE_BUILD_TYPE=Release", "-DA2A_ENABLE_TESTING=ON"]
+    if "postgres" in config.store_backends:
+        configure.append("-DA2A_ENABLE_POSTGRES_STORE=ON")
+    subprocess.run(configure, check=True)
+    subprocess.run(["cmake", "--build", str(build_dir), "--target", DRIVER_NAME, "-j", str(os.cpu_count() or 2)], check=True)
+    if not driver.exists():
+        raise ValueError(f"performance driver was not produced at {driver}")
+    return driver
+
+
+def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]:
+    driver = ensure_driver(config)
+    completed = subprocess.run([
+        str(driver),
+        "--transport", transport,
+        "--store-backend", store_backend,
+        "--requests", str(config.requests),
+        "--concurrency", str(concurrency),
+        "--warmup-seconds", str(config.warmup_seconds),
+        "--duration-seconds", str(config.duration_seconds),
+    ], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False)
+    if completed.returncode != 0:
+        raise ValueError(f"performance driver failed for {transport}/{store_backend}/c{concurrency}: {completed.stderr.strip()}")
+    payload = json.loads(completed.stdout)
+    if not isinstance(payload, list):
+        raise ValueError("performance driver returned malformed output")
+    return payload
 
 def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]:
     items = tuple(item.strip() for item in value.split(",") if item.strip())
@@ -156,62 +159,6 @@ def parse_args(argv: list[str]) -> RunnerConfig:
     )
 
 
-def percentile(sorted_values: list[float], percentile_rank: float) -> float:
-    if not sorted_values:
-        return 0.0
-    index = min(len(sorted_values) - 1, round((percentile_rank / 100.0) * (len(sorted_values) - 1)))
-    return sorted_values[index]
-
-
-def run_one(config: RunnerConfig, scenario: str, transport: str, store_backend: str, concurrency: int) -> dict[str, object]:
-    state = ScenarioState()
-    end_warmup = time.perf_counter() + config.warmup_seconds
-    warmup_index = 0
-    while time.perf_counter() < end_warmup:
-        state.execute(scenario, warmup_index)
-        warmup_index += 1
-
-    latencies: list[float] = []
-    errors = 0
-    started = time.perf_counter()
-
-    def operation(index: int) -> float:
-        op_started = time.perf_counter_ns()
-        if not state.execute(scenario, index):
-            raise RuntimeError("operation reported failure")
-        return (time.perf_counter_ns() - op_started) / NANOSECONDS_PER_MILLISECOND
-
-    with ThreadPoolExecutor(max_workers=concurrency) as executor:
-        futures = [executor.submit(operation, index) for index in range(config.requests)]
-        for future in as_completed(futures):
-            try:
-                latencies.append(future.result())
-            except Exception:
-                errors += 1
-    elapsed = max(time.perf_counter() - started, sys.float_info.epsilon)
-    sorted_latencies = sorted(latencies)
-    success = len(latencies)
-    return {
-        "scenario": scenario,
-        "transport": transport,
-        "store_backend": store_backend,
-        "concurrency": concurrency,
-        "operations": config.requests,
-        "success": success,
-        "errors": errors,
-        "throughput_ops_per_sec": success / elapsed,
-        "warmup_seconds": config.warmup_seconds,
-        "duration_seconds": config.duration_seconds,
-        "latency_ms": {
-            "p50": percentile(sorted_latencies, 50.0),
-            "p90": percentile(sorted_latencies, 90.0),
-            "p95": percentile(sorted_latencies, 95.0),
-            "p99": percentile(sorted_latencies, 99.0),
-            "max": max(sorted_latencies) if sorted_latencies else 0.0,
-        },
-    }
-
-
 def commit_sha() -> str:
     try:
         return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
@@ -329,7 +276,8 @@ def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object]
 def main(argv: list[str]) -> int:
     try:
         config = parse_args(argv)
-        results = [run_one(config, scenario, transport, store_backend, concurrency) for scenario in SCENARIOS for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels]
+        results = [result for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels for result in run_driver(config, transport, store_backend, concurrency)]
+        results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["transport"]), int(result["concurrency"])))
         write_reports(results, config)
         if any(result["errors"] for result in results):
             return 2