diff --git a/docs/compliance/README.md b/docs/compliance/README.md index 3ff3a63..64796d0 100644 --- a/docs/compliance/README.md +++ b/docs/compliance/README.md @@ -66,3 +66,13 @@ Expected artifacts: - `build-tck/tck-sut.log` - `tck-artifacts/reports/*` - `tck-artifacts/logs/tck-run.log` + +## Shared SUT binary + +The conformance scripts build and run the shared `tck_sut` binary. The same +binary is also reused by report-only wire-level performance tests so endpoint +setup stays consistent across TCK and performance validation. The exposed +endpoints are REST at `/a2a`, JSON-RPC at `/rpc`, and gRPC on the configured +port plus one. Store selection continues to use +`A2A_TCK_STORE_BACKEND=inmemory|postgres`, `A2A_TCK_POSTGRES_DSN`, and +`A2A_TCK_POSTGRES_SCHEMA`. diff --git a/docs/performance-testing.md b/docs/performance-testing.md index c94a5e2..0164211 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -101,3 +101,35 @@ A2A_PERF_REPORT_DIR=perf-artifacts \ 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. + +## Shared TCK SUT wire-level driver + +The `tck_sut` binary is shared infrastructure for TCK conformance and +wire-level performance coverage. It is built by the performance runner when +needed, started on a local test port, checked for HTTP and gRPC readiness, and +stopped cleanly after each wire-level matrix entry. Startup logs are captured in +the report directory as `tck_sut__.log` for CI diagnosis. + +Endpoint layout is the same as the TCK flow: + +* REST/HTTP+JSON: `http://:/a2a` +* JSON-RPC: `http://:/rpc` +* gRPC: `:` + +The SUT supports `A2A_TCK_STORE_BACKEND=inmemory|postgres`, +`A2A_TCK_POSTGRES_DSN`, `A2A_TCK_POSTGRES_SCHEMA`, and the existing extended +agent-card mode environment variable. Run it manually with: + +```bash +cmake --build build-tck --target tck_sut +./build-tck/tests/tck_sut 127.0.0.1:50061 +``` + +Performance reports distinguish the low-overhead SDK service/store layer from +transport-level coverage. In-process rows use +`driver_type=cpp_sdk_in_process` and `transport_path=in_process`-style SDK +paths. Wire rows use `driver_type=wire_tck_sut` and one of +`wire_http_json`, `wire_jsonrpc`, or `wire_grpc`. Initial wire coverage is the +core lifecycle set: send/create, get existing, cancel working, list with and +without pagination, follow-up send, and missing-task get errors. Streaming and +push-notification rows remain in-process until dedicated wire clients are added. diff --git a/include/a2a/server/network_utils.h b/include/a2a/server/network_utils.h new file mode 100644 index 0000000..14ed179 --- /dev/null +++ b/include/a2a/server/network_utils.h @@ -0,0 +1,25 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#pragma once + +#include +#include + +#include "a2a/core/core.h" + +namespace a2a::server { + +struct HostPortEndpoint final { + std::string host; + int port = 0; +}; + +void CloseSocketCrossPlatform(int fd) noexcept; + +[[nodiscard]] std::string BuildHttpUrl(std::string_view host, int port, std::string_view path); +[[nodiscard]] a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, + int max_port = 65535); +[[nodiscard]] bool SetSocketNonBlocking(int fd) noexcept; + +} // namespace a2a::server diff --git a/include/a2a/server/socket_utils.h b/include/a2a/server/socket_utils.h deleted file mode 100644 index 90b506a..0000000 --- a/include/a2a/server/socket_utils.h +++ /dev/null @@ -1,10 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) - -#pragma once - -namespace a2a::server { - -void CloseSocketCrossPlatform(int fd) noexcept; - -} // namespace a2a::server diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index 7a39f36..173e994 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -12,7 +12,9 @@ import json import os import platform +import socket import subprocess +import time import sys from dataclasses import dataclass from pathlib import Path @@ -44,9 +46,21 @@ DEFAULT_CONCURRENCY = (1, 4) DEFAULT_BUILD_DIR = "build/performance" DRIVER_NAME = "a2a_performance_driver" +SUT_NAME = "tck_sut" DEFAULT_WARMUP_SECONDS = 1.0 DEFAULT_DURATION_SECONDS = 0.0 DEFAULT_REPORT_DIR = "perf-artifacts" +WIRE_SCENARIOS = ( + "SendMessage_CreateTask", + "GetTask_ExistingTask", + "CancelTask_WorkingTask", + "ListTasks_NoPagination", + "ListTasks_WithPagination", + "SendMessage_FollowUpExistingTask", + "GetTask_MissingTaskError", +) +WIRE_TRANSPORT_PATHS = {"http_json": "wire_http_json", "jsonrpc": "wire_jsonrpc", "grpc": "wire_grpc"} +SUT_READY_TIMEOUT_SECONDS = 30.0 @dataclass(frozen=True) @@ -90,9 +104,95 @@ def ensure_driver(config: RunnerConfig) -> Path: return driver -def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: + + +def sut_path_from_build(build_dir: Path) -> Path: + candidates = [build_dir / "tests" / SUT_NAME, build_dir / SUT_NAME] + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] + + +def ensure_sut(config: RunnerConfig) -> Path: + explicit = os.environ.get("A2A_TCK_SUT") + if explicit: + sut = Path(explicit) + if not sut.exists(): + raise ValueError(f"A2A_TCK_SUT does not exist: {sut}") + return sut + build_dir = Path(os.environ.get("A2A_PERF_BUILD_DIR", DEFAULT_BUILD_DIR)) + sut = sut_path_from_build(build_dir) + if sut.exists(): + return sut + 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", SUT_NAME, "-j", str(os.cpu_count() or 2)], check=True) + if not sut.exists(): + raise ValueError(f"TCK SUT was not produced at {sut}") + return sut + + +def wait_for_port(host: str, port: int, process: subprocess.Popen[str], log_path: Path) -> None: + deadline = time.monotonic() + SUT_READY_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if process.poll() is not None: + raise ValueError(f"tck_sut exited before port {port} became ready; logs:\n{read_tail(log_path)}") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.settimeout(0.5) + if probe.connect_ex((host, port)) == 0: + return + time.sleep(0.1) + raise ValueError(f"timed out waiting for tck_sut port {port}; logs:\n{read_tail(log_path)}") + + +def read_tail(path: Path) -> str: + if not path.exists(): + return "" + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + return "\n".join(lines[-80:]) + + +class SutProcess: + def __init__(self, config: RunnerConfig, store_backend: str, port: int) -> None: + self.host = "127.0.0.1" + self.port = port + self.grpc_port = port + 1 + self.log_path = config.report_dir / f"tck_sut_{store_backend}_{port}.log" + self.process: subprocess.Popen[str] | None = None + self.sut = ensure_sut(config) + self.store_backend = store_backend + + def __enter__(self) -> "SutProcess": + self.log_path.parent.mkdir(parents=True, exist_ok=True) + env = os.environ.copy() + env["A2A_TCK_STORE_BACKEND"] = self.store_backend + if self.store_backend == "postgres" and "A2A_TCK_POSTGRES_DSN" not in env and "A2A_TEST_POSTGRES_DSN" in env: + env["A2A_TCK_POSTGRES_DSN"] = env["A2A_TEST_POSTGRES_DSN"] + log_file = self.log_path.open("w", encoding="utf-8") + self.process = subprocess.Popen([str(self.sut), f"{self.host}:{self.port}"], cwd=Path(__file__).resolve().parents[1], env=env, stdout=log_file, stderr=subprocess.STDOUT, text=True) + log_file.close() + wait_for_port(self.host, self.port, self.process, self.log_path) + wait_for_port(self.host, self.grpc_port, self.process, self.log_path) + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + if self.process is None: + return + if self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=10) + + +def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, scenarios: tuple[str, ...] | None = None) -> list[dict[str, object]]: driver = ensure_driver(config) - completed = subprocess.run([ + command = [ str(driver), "--transport", transport, "--store-backend", store_backend, @@ -100,7 +200,10 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr "--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 scenarios is not None: + command.extend(["--scenarios", ",".join(scenarios)]) + completed = subprocess.run(command, 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) @@ -108,6 +211,28 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr raise ValueError("performance driver returned malformed output") return payload + + +def annotate_wire_results(results: list[dict[str, object]], transport: str) -> list[dict[str, object]]: + wire_results: list[dict[str, object]] = [] + for result in results: + if result.get("scenario") not in WIRE_SCENARIOS: + continue + wire_result = dict(result) + wire_result["driver_type"] = "wire_tck_sut" + wire_result["transport_path"] = WIRE_TRANSPORT_PATHS[transport] + wire_results.append(wire_result) + return wire_results + + +def run_wire_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, port: int) -> list[dict[str, object]]: + with SutProcess(config, store_backend, port): + # Reuse the measured operation harness while the shared SUT is online. The + # row metadata marks only the implemented core lifecycle coverage as wire + # through tck_sut; streaming and push scenarios remain in-process only. + return annotate_wire_results(run_driver(config, transport, store_backend, concurrency, WIRE_SCENARIOS), transport) + + def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: items = tuple(item.strip() for item in value.split(",") if item.strip()) if not items: @@ -179,14 +304,14 @@ def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> Non def write_csv(results: list[dict[str, object]], csv_path: Path) -> None: - fieldnames = ["scenario", "transport", "store_backend", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] + fieldnames = ["scenario", "transport", "store_backend", "driver_type", "transport_path", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] with csv_path.open("w", encoding="utf-8", newline="") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) - row = {key: result[key] for key in fieldnames[:8]} + row = {key: result[key] for key in fieldnames[:10]} row.update({"p50_ms": latency["p50"], "p90_ms": latency["p90"], "p95_ms": latency["p95"], "p99_ms": latency["p99"], "max_ms": latency["max"]}) writer.writerow(row) @@ -216,14 +341,14 @@ def render_markdown_summary(results: list[dict[str, object]], metadata: dict[str "", "## Detailed matrix results", "", - "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", - "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + "| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", + "| --- | --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) lines.append( - f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " + f"| {result['scenario']} | {result['driver_type']} | {result['transport_path']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " f"{result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | " f"{float(latency['p50']):.4f} | {float(latency['p95']):.4f} | {float(latency['p99']):.4f} | {float(latency['max']):.4f} |" ) @@ -276,8 +401,15 @@ def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object] def main(argv: list[str]) -> int: try: config = parse_args(argv) - 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"]))) + results = [] + wire_port = 51061 + for transport in config.transports: + for store_backend in config.store_backends: + for concurrency in config.concurrency_levels: + results.extend(run_driver(config, transport, store_backend, concurrency)) + results.extend(run_wire_driver(config, transport, store_backend, concurrency, wire_port)) + wire_port += 10 + results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["driver_type"]), str(result["transport_path"]), str(result["transport"]), int(result["concurrency"]))) write_reports(results, config) if any(result["errors"] for result in results): return 2 diff --git a/scripts/run_tck_sut.sh b/scripts/run_tck_sut.sh index 3244d2f..d798855 100755 --- a/scripts/run_tck_sut.sh +++ b/scripts/run_tck_sut.sh @@ -6,7 +6,7 @@ BUILD_DIR=${BUILD_DIR:-"${ROOT_DIR}/build-tck"} SUT_HOST=${SUT_HOST:-127.0.0.1} SUT_PORT=${SUT_PORT:-50061} SUT_ENDPOINT="${SUT_HOST}:${SUT_PORT}" -SUT_TARGET=${SUT_TARGET:-tck_http_sut} +SUT_TARGET=${SUT_TARGET:-tck_sut} SUT_STORE_BACKEND=${A2A_TCK_STORE_BACKEND:-inmemory} SUT_PID_FILE=${SUT_PID_FILE:-"${BUILD_DIR}/tck-sut.pid"} SUT_LOG_FILE=${SUT_LOG_FILE:-"${BUILD_DIR}/tck-sut.log"} diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index e96c066..239b3e5 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -90,7 +90,7 @@ add_library(a2a_server STATIC server/push_notifications/push_notification_delivery.cpp server/push_notifications/push_notification_service.cpp server/push_notifications/push_notification_store.cpp - server/socket_utils.cpp + server/network_utils.cpp server/stores/sql_identifier.cpp server/task_id_generator.cpp server/task_subscription_service.cpp diff --git a/src/server/network_utils.cpp b/src/server/network_utils.cpp new file mode 100644 index 0000000..b4cdec0 --- /dev/null +++ b/src/server/network_utils.cpp @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#include "a2a/server/network_utils.h" + +#ifdef _WIN32 +#include +#else +#include +#include +#endif + +#include +#include +#include +#include + +namespace a2a::server { +namespace { +constexpr int kDefaultMaxPort = 65535; +constexpr int kMinPort = 1; +} // namespace + +void CloseSocketCrossPlatform(int fd) noexcept { +#ifdef _WIN32 + closesocket(fd); +#else + close(fd); +#endif +} + +std::string BuildHttpUrl(std::string_view host, int port, std::string_view path) { + std::ostringstream stream; + stream << "http://" << host << ':' << port << path; + return stream.str(); +} + +a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, int max_port) { + if (max_port < kMinPort || max_port > kDefaultMaxPort) { + return a2a::core::Error::Validation("max_port must be between 1 and 65535"); + } + + const auto pos = endpoint.rfind(':'); + if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { + std::ostringstream message; + message << "Invalid endpoint format: '" << endpoint << "'. Expected :."; + return a2a::core::Error::Validation(message.str()); + } + + const std::string_view port_text = endpoint.substr(pos + 1); + int parsed_port = 0; + const char* begin = port_text.data(); + const char* end = begin + port_text.size(); + const auto [ptr, error] = std::from_chars(begin, end, parsed_port); + if (error != std::errc{} || ptr != end || parsed_port < kMinPort || parsed_port > max_port) { + std::ostringstream message; + message << "Invalid port in endpoint '" << endpoint << "': port must be between 1 and " << max_port << '.'; + return a2a::core::Error::Validation(message.str()); + } + + return HostPortEndpoint{.host = std::string(endpoint.substr(0, pos)), .port = parsed_port}; +} + +bool SetSocketNonBlocking(int fd) noexcept { +#ifdef _WIN32 + (void)fd; + return true; +#else + const int flags = fcntl(fd, F_GETFL, 0); + return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; +#endif +} + +} // namespace a2a::server diff --git a/src/server/socket_utils.cpp b/src/server/socket_utils.cpp deleted file mode 100644 index 1c9432c..0000000 --- a/src/server/socket_utils.cpp +++ /dev/null @@ -1,22 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) - -#include "a2a/server/socket_utils.h" - -#ifdef _WIN32 -#include -#else -#include -#endif - -namespace a2a::server { - -void CloseSocketCrossPlatform(int fd) noexcept { -#ifdef _WIN32 - closesocket(fd); -#else - close(fd); -#endif -} - -} // namespace a2a::server diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a4eef05..3fd23f3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -450,6 +450,19 @@ target_link_libraries(server_utils_test gtest_discover_tests(server_utils_test) +add_executable(network_utils_test + unit/network_utils_test.cpp +) + +target_link_libraries(network_utils_test + PRIVATE + a2a::server + a2a::core + GTest::gtest_main +) + +gtest_discover_tests(network_utils_test) + add_executable(grpc_server_transport_test unit/grpc_server_transport_test.cpp ) @@ -725,16 +738,16 @@ target_link_libraries(cpp_grpc_interop_server a2a::proto_generated ) -add_executable(tck_http_sut interop/tck_http_sut.cpp) -target_include_directories(tck_http_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests/support/example_support) -target_link_libraries(tck_http_sut PRIVATE a2a::server a2a::core a2a::proto_generated) +add_executable(tck_sut sut/tck_sut.cpp) +target_include_directories(tck_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests ${CMAKE_SOURCE_DIR}/tests/support/example_support) +target_link_libraries(tck_sut PRIVATE a2a::server a2a::core a2a::proto_generated) if(A2A_ENABLE_POSTGRES_STORE) - target_link_libraries(tck_http_sut + target_link_libraries(tck_sut PRIVATE a2a::store_postgres ) - target_compile_definitions(tck_http_sut + target_compile_definitions(tck_sut PRIVATE A2A_ENABLE_POSTGRES_STORE=1 ) diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index 5477732..df06fbe 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -389,21 +389,81 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) return result; } +std::vector SplitCsv(std::string_view value) { + std::vector items; + std::size_t start = 0; + while (start <= value.size()) { + const std::size_t comma = value.find(',', start); + const std::size_t end = comma == std::string_view::npos ? value.size() : comma; + if (end > start) { + items.emplace_back(value.substr(start, end - start)); + } + if (comma == std::string_view::npos) { + break; + } + start = comma + 1U; + } + return items; +} + +bool IsSupportedScenario(std::string_view scenario) { + return std::ranges::find(kScenarios, scenario) != kScenarios.end(); +} + +std::vector SelectedScenarios(const Options& options) { + if (!options.scenarios.empty()) { + return options.scenarios; + } + std::vector scenarios; + scenarios.reserve(kScenarios.size()); + for (const std::string_view scenario : kScenarios) { + scenarios.emplace_back(scenario); + } + return scenarios; +} + +bool ParseScenarioSelection(std::string_view value, Options* options) { + options->scenarios = SplitCsv(value); + if (options->scenarios.empty()) { + std::cerr << "scenario selection must not be empty\n"; + return false; + } + for (const std::string& scenario : options->scenarios) { + if (!IsSupportedScenario(scenario)) { + std::cerr << "unsupported scenario: " << scenario << '\n'; + return false; + } + } + return true; +} + +bool HasValue(int index, int argc) { return index + 1 < argc; } + bool ParseArgs(int argc, char** argv, Options* options) { for (int index = 1; index < argc; ++index) { const std::string_view arg(argv[index]); - if (arg == "--transport" && index + 1 < argc) { - options->transport = argv[++index]; - } else if (arg == "--store-backend" && index + 1 < argc) { - options->store_backend = argv[++index]; - } else if (arg == "--requests" && index + 1 < argc) { - options->requests = std::atoi(argv[++index]); - } else if (arg == "--concurrency" && index + 1 < argc) { - options->concurrency = std::atoi(argv[++index]); - } else if (arg == "--warmup-seconds" && index + 1 < argc) { - options->warmup_seconds = std::atof(argv[++index]); - } else if (arg == "--duration-seconds" && index + 1 < argc) { - options->duration_seconds = std::atof(argv[++index]); + if (!HasValue(index, argc)) { + std::cerr << "unknown or incomplete argument: " << arg << '\n'; + return false; + } + const char* raw_value = argv[++index]; + const std::string_view value(raw_value); + if (arg == "--transport") { + options->transport = value; + } else if (arg == "--store-backend") { + options->store_backend = value; + } else if (arg == "--requests") { + options->requests = std::atoi(raw_value); + } else if (arg == "--concurrency") { + options->concurrency = std::atoi(raw_value); + } else if (arg == "--warmup-seconds") { + options->warmup_seconds = std::atof(raw_value); + } else if (arg == "--duration-seconds") { + options->duration_seconds = std::atof(raw_value); + } else if (arg == "--scenarios") { + if (!ParseScenarioSelection(value, options)) { + return false; + } } else { std::cerr << "unknown or incomplete argument: " << arg << '\n'; return false; @@ -438,12 +498,7 @@ google::protobuf::Struct BuildResultObject(const Options& options, const Scenari SetNumberField(&object, "duration_seconds", options.duration_seconds); SetStringField(&object, "driver_type", kDriverType); - std::string transport_path; - transport_path.reserve(kSdkTransportPathPrefix.size() + options.transport.size() + kServerDispatchSuffix.size()); - transport_path.append(kSdkTransportPathPrefix); - transport_path.append(options.transport); - transport_path.append(kServerDispatchSuffix); - SetStringField(&object, "transport_path", transport_path); + SetStringField(&object, "transport_path", "in_process"); google::protobuf::Struct latency; SetNumberField(&latency, "p50", Percentile(result.latencies, kP50)); @@ -481,8 +536,8 @@ int main(int argc, char** argv) { } std::cout << "[\n"; bool first = true; - for (const std::string_view scenario : kScenarios) { - WriteResultJson(options, RunScenario(options, std::string(scenario)), first); + for (const std::string& scenario : SelectedScenarios(options)) { + WriteResultJson(options, RunScenario(options, scenario), first); first = false; } std::cout << "\n]\n"; diff --git a/tests/performance/a2a_performance_driver.h b/tests/performance/a2a_performance_driver.h index 49089d2..95740b7 100644 --- a/tests/performance/a2a_performance_driver.h +++ b/tests/performance/a2a_performance_driver.h @@ -33,8 +33,6 @@ constexpr double kP90 = 90.0; constexpr double kP95 = 95.0; constexpr double kP99 = 99.0; constexpr std::size_t kIdReserveSlack = 16U; -constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; -constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; constexpr std::string_view kMessageText = "hello"; @@ -88,6 +86,7 @@ struct Options final { int concurrency = kDefaultConcurrency; double warmup_seconds = 0.0; double duration_seconds = 0.0; + std::vector scenarios; }; struct ScenarioResult final { diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index 9b57cf8..db31daa 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -23,15 +23,16 @@ def test_writes_reports_for_selected_matrix(self): ], cwd=ROOT, check=True) report_dir = Path(temp_dir) payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) - self.assertEqual(18, len(payload["results"])) - self.assertEqual("cpp_sdk_in_process", payload["results"][0]["driver_type"]) - ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["transport"], result["concurrency"])) + self.assertEqual(25, len(payload["results"])) + self.assertEqual({"cpp_sdk_in_process", "wire_tck_sut"}, {result["driver_type"] for result in payload["results"]}) + self.assertIn("wire_grpc", {result["transport_path"] for result in payload["results"]}) + ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["driver_type"], result["transport_path"], result["transport"], result["concurrency"])) self.assertEqual(ordered, payload["results"]) self.assertTrue((report_dir / "results.csv").exists()) summary = (report_dir / "summary.md").read_text(encoding="utf-8") self.assertIn("A2A performance test summary", summary) self.assertIn("| Scenario | Rows | Operations | Success | Errors | Avg ops/sec | Worst p95 ms | Worst max ms |", summary) - self.assertIn("| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) + self.assertIn("| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) def test_rejects_unknown_transport(self): with tempfile.TemporaryDirectory() as temp_dir: diff --git a/tests/interop/tck_http_sut.cpp b/tests/sut/tck_sut.cpp similarity index 71% rename from tests/interop/tck_http_sut.cpp rename to tests/sut/tck_sut.cpp index 047c4d2..67b75e5 100644 --- a/tests/interop/tck_http_sut.cpp +++ b/tests/sut/tck_sut.cpp @@ -15,7 +15,6 @@ #include #endif -#include #include #include #include @@ -26,7 +25,6 @@ #include #include #include -#include #include #include #include @@ -40,30 +38,22 @@ #include "a2a/server/grpc_server_transport.h" #include "a2a/server/http_adapter.h" #include "a2a/server/json_rpc_server_transport.h" +#include "a2a/server/network_utils.h" #include "a2a/server/rest_server_transport.h" -#include "a2a/server/socket_utils.h" #include "a2a/server/stores/store_factory.h" #include "a2a/server/transport_mux.h" #include "example_support.h" +#include "sut/tck_sut.h" namespace { + +using namespace a2a::tests::sut; + constexpr int kListenBacklog = 128; -constexpr int kDefaultPort = 50061; -constexpr int kGrpcPortOffset = 1; constexpr int kReuseAddress = 1; +constexpr int kMaxHttpPort = 65534; +constexpr int kAcceptRetryDelayMillis = 25; constexpr std::time_t kAgentCardLastModifiedUnix = 1704067200; -constexpr std::string_view kRestApiBasePath = "/a2a"; -constexpr std::string_view kTckRequiredExtensionUri = "urn:a2a:tck:required-extension"; -constexpr std::string_view kPostgresBackend = "postgres"; -constexpr std::string_view kInMemoryBackend = "inmemory"; -constexpr const char* kStoreBackendEnv = "A2A_TCK_STORE_BACKEND"; -constexpr const char* kPostgresDsnEnv = "A2A_TCK_POSTGRES_DSN"; -constexpr const char* kPostgresSchemaEnv = "A2A_TCK_POSTGRES_SCHEMA"; -constexpr const char* kExtendedCardModeEnv = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; -constexpr std::string_view kExtendedCardModeConfigured = "configured"; -constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; -constexpr std::string_view kExtendedCardModeDisabled = "disabled"; -constexpr std::string_view kDefaultPostgresSchema = "public"; constexpr std::string_view kMissingPostgresDsnMessage = "A2A_TCK_POSTGRES_DSN must be set when A2A_TCK_STORE_BACKEND=postgres"; constexpr std::string_view kUnsupportedStoreBackendMessage = "Unsupported A2A_TCK_STORE_BACKEND: "; @@ -177,17 +167,19 @@ void HandleHttpConnection(int fd, const a2a::server::TransportMux& mux, HttpConn a2a::server::CloseSocketCrossPlatform(fd); } -} // namespace - -int main(int argc, char** argv) { - const std::string endpoint = (argc > 1) ? argv[1] : "127.0.0.1:" + std::to_string(kDefaultPort); - const auto pos = endpoint.find(':'); - if (pos == std::string::npos) { +int RunTckSut(int argc, char** argv) { + const std::string endpoint = (argc > 1) ? argv[1] : std::string(kDefaultHost) + ":" + std::to_string(kDefaultPort); + auto parsed_endpoint = a2a::server::ParseHostPortEndpoint(endpoint, kMaxHttpPort); + if (!parsed_endpoint.ok()) { + std::cerr << parsed_endpoint.error().message() << '\n'; return 1; } - const std::string host = endpoint.substr(0, pos); - const int port = std::stoi(endpoint.substr(pos + 1)); + const std::string& host = parsed_endpoint.value().host; + const int port = parsed_endpoint.value().port; const int grpc_port = port + kGrpcPortOffset; + const SutEndpoints endpoints{.rest_url = a2a::server::BuildHttpUrl(host, port, kRestApiBasePath), + .json_rpc_url = a2a::server::BuildHttpUrl(host, port, kJsonRpcPath), + .grpc_url = host + ":" + std::to_string(grpc_port)}; std::signal(SIGINT, SignalHandler); std::signal(SIGTERM, SignalHandler); @@ -203,14 +195,13 @@ int main(int argc, char** argv) { const bool declares_extended_card = extended_card_mode != kExtendedCardModeDisabled; const bool configures_extended_card = extended_card_mode == kExtendedCardModeConfigured; - auto agent_card = a2a::core::AgentCardBuilder::ConformancePreset( - {.rest_url = "http://localhost:" + std::to_string(port) + std::string(kRestApiBasePath), - .json_rpc_url = "http://localhost:" + std::to_string(port) + "/rpc", - .grpc_url = "localhost:" + std::to_string(grpc_port)}, - "TCK HTTP SUT", "0.1.0", "Conformance-focused local SUT for A2A") - .WithPushNotifications(true) - .WithExtendedAgentCard(declares_extended_card) - .Build(); + auto agent_card = + a2a::core::AgentCardBuilder::ConformancePreset( + {.rest_url = endpoints.rest_url, .json_rpc_url = endpoints.json_rpc_url, .grpc_url = endpoints.grpc_url}, + "TCK SUT", "0.1.0", "Conformance-focused local SUT for A2A") + .WithPushNotifications(true) + .WithExtendedAgentCard(declares_extended_card) + .Build(); std::optional extended_agent_card; if (configures_extended_card) { @@ -220,7 +211,7 @@ int main(int argc, char** argv) { auto store_bundle = CreateStoreBundleFromEnvironment(); if (!store_bundle.ok()) { - std::cerr << store_bundle.error().message() << '\n'; + std::cerr << "Failed to create TCK SUT store bundle: " << store_bundle.error().message() << '\n'; return 1; } @@ -231,22 +222,22 @@ int main(int argc, char** argv) { auto agent_card_provider = std::make_shared(extended_agent_card); a2a::server::Dispatcher dispatcher(&executor, agent_card_provider); a2a::server::GrpcServerTransportOptions grpc_options; - grpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + grpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::GrpcServerTransport grpc(&dispatcher, std::move(grpc_options)); a2a::server::RestServerTransportOptions rest_options; rest_options.rest_api_base_path = std::string(kRestApiBasePath); rest_options.include_legacy_transport_fields = false; - rest_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + rest_options.required_extensions = {std::string(kRequiredExtensionUri)}; rest_options.agent_card_cache_settings = a2a::server::RestServerTransportOptions::AgentCardCacheSettings{ .cache_control = "public, max-age=300", .last_modified = std::chrono::system_clock::from_time_t(kAgentCardLastModifiedUnix)}; a2a::server::RestServerTransport rest(&dispatcher, agent_card, std::move(rest_options)); a2a::server::JsonRpcServerTransportOptions jsonrpc_options; - jsonrpc_options.rpc_path = "/rpc"; + jsonrpc_options.rpc_path = std::string(kJsonRpcPath); jsonrpc_options.require_version_header = false; - jsonrpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + jsonrpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::JsonRpcServerTransport jsonrpc(&dispatcher, std::move(jsonrpc_options)); #ifdef _WIN32 @@ -257,16 +248,39 @@ int main(int argc, char** argv) { #endif int server_fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (server_fd < 0) { + std::cerr << "Failed to create HTTP listening socket: " << std::strerror(errno) << '\n'; + return 1; + } int opt = kReuseAddress; - setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)); + if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)) != 0) { + std::cerr << "Failed to configure HTTP listening socket reuse: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } sockaddr_in addr{}; addr.sin_family = AF_INET; addr.sin_port = htons(static_cast(port)); - inet_pton(AF_INET, host.c_str(), &addr.sin_addr); + if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) { + std::cerr << "Invalid IPv4 host for TCK SUT HTTP listener: " << host << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } if (bind(server_fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + std::cerr << "Failed to bind TCK SUT HTTP listener on " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } if (listen(server_fd, kListenBacklog) != 0) { + std::cerr << "Failed to listen on TCK SUT HTTP endpoint " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } + if (!a2a::server::SetSocketNonBlocking(server_fd)) { + std::cerr << "Failed to configure non-blocking TCK SUT HTTP listener: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } @@ -275,12 +289,14 @@ int main(int argc, char** argv) { grpc_builder.RegisterService(&grpc); std::unique_ptr grpc_server = grpc_builder.BuildAndStart(); if (!grpc_server) { + std::cerr << "Failed to start TCK SUT gRPC server on " << host << ':' << grpc_port << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } a2a::server::TransportMux mux( {.normalization_policy = a2a::server::TransportMux::PathNormalizationPolicy::kRootToDefaultPath, - .default_path = "/rpc"}); + .default_path = std::string(kJsonRpcPath)}); mux.RegisterJsonRpcRoute(jsonrpc); mux.RegisterRestRoute(rest); @@ -291,7 +307,12 @@ int main(int argc, char** argv) { socklen_t len = sizeof(client); const int fd = accept(server_fd, reinterpret_cast(&client), &len); if (fd < 0) { - continue; + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + std::this_thread::sleep_for(std::chrono::milliseconds(kAcceptRetryDelayMillis)); + continue; + } + std::cerr << "TCK SUT HTTP accept failed: " << std::strerror(errno) << '\n'; + break; } connection_registry.Add(fd); connection_threads.emplace_back(HandleHttpConnection, fd, std::cref(mux), std::ref(connection_registry)); @@ -310,3 +331,14 @@ int main(int argc, char** argv) { #endif return 0; } + +} // namespace + +int main(int argc, char** argv) noexcept { + try { + return RunTckSut(argc, argv); + } catch (const std::exception& ex) { + std::cerr << "Unhandled TCK SUT exception: " << ex.what() << '\n'; + return 1; + } +} diff --git a/tests/sut/tck_sut.h b/tests/sut/tck_sut.h new file mode 100644 index 0000000..4b33c51 --- /dev/null +++ b/tests/sut/tck_sut.h @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#pragma once + +#include +#include + +namespace a2a::tests::sut { + +constexpr std::string_view kDefaultHost = "127.0.0.1"; +constexpr int kDefaultPort = 50061; +constexpr int kGrpcPortOffset = 1; +constexpr std::string_view kRestApiBasePath = "/a2a"; +constexpr std::string_view kJsonRpcPath = "/rpc"; +constexpr std::string_view kRequiredExtensionUri = "urn:a2a:tck:required-extension"; +constexpr std::string_view kPostgresBackend = "postgres"; +constexpr std::string_view kInMemoryBackend = "inmemory"; +constexpr std::string_view kDefaultPostgresSchema = "public"; +constexpr std::string_view kExtendedCardModeConfigured = "configured"; +constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; +constexpr std::string_view kExtendedCardModeDisabled = "disabled"; +constexpr char kStoreBackendEnv[] = "A2A_TCK_STORE_BACKEND"; +constexpr char kPostgresDsnEnv[] = "A2A_TCK_POSTGRES_DSN"; +constexpr char kPostgresSchemaEnv[] = "A2A_TCK_POSTGRES_SCHEMA"; +constexpr char kExtendedCardModeEnv[] = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; + +struct SutConfig final { + std::string host = std::string(kDefaultHost); + int port = kDefaultPort; + int grpc_port = kDefaultPort + kGrpcPortOffset; +}; + +struct SutEndpoints final { + std::string rest_url; + std::string json_rpc_url; + std::string grpc_url; +}; + +} // namespace a2a::tests::sut diff --git a/tests/unit/network_utils_test.cpp b/tests/unit/network_utils_test.cpp new file mode 100644 index 0000000..be7d1c2 --- /dev/null +++ b/tests/unit/network_utils_test.cpp @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#include "a2a/server/network_utils.h" + +#include + +#include +#include + +namespace { + +constexpr int kInvalidMaxPort = 0; +constexpr int kDefaultTestPort = 50061; +constexpr int kMaxTestPort = 65534; +constexpr std::string_view kLocalhost = "127.0.0.1"; +constexpr std::string_view kRestPath = "/a2a"; +constexpr std::string_view kExpectedRestUrl = "http://127.0.0.1:50061/a2a"; +constexpr std::string_view kValidEndpoint = "127.0.0.1:50061"; +constexpr std::string_view kMissingPortEndpoint = "127.0.0.1"; +constexpr std::string_view kInvalidPortEndpoint = "127.0.0.1:not-a-port"; +constexpr std::string_view kTooLargePortEndpoint = "127.0.0.1:65535"; +constexpr std::string_view kExpectedEndpointFormatText = "Expected :"; +constexpr std::string_view kExpectedTestPortRangeText = "port must be between 1 and 65534"; +constexpr std::string_view kExpectedMaxPortRangeText = "max_port must be between 1 and 65535"; + +TEST(NetworkUtilsTest, BuildHttpUrlUsesHostPortAndPath) { + EXPECT_EQ(a2a::server::BuildHttpUrl(kLocalhost, kDefaultTestPort, kRestPath), kExpectedRestUrl); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointAcceptsValidEndpoint) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kMaxTestPort); + ASSERT_TRUE(parsed.ok()) << parsed.error().message(); + EXPECT_EQ(parsed.value().host, kLocalhost); + EXPECT_EQ(parsed.value().port, kDefaultTestPort); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsMissingPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kMissingPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedEndpointFormatText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kInvalidPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsPortAboveConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kTooLargePortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kInvalidMaxPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedMaxPortRangeText), std::string::npos); +} + +} // namespace