"""ASV benchmarks for async command execution and workers. Measures the performance of: - AsyncJob model creation and validation overhead - AsyncJob state machine transitions - InMemoryJobStore CRUD operations - AsyncWorkerConfig instantiation - WorkerHealthReport lifecycle - CancellationToken operations - Payload serialization/deserialization - AsyncWorker job pickup and execution """ from __future__ import annotations import importlib import sys from pathlib import Path # Ensure the local *source* tree is importable even when ASV has an # older build of the package installed. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) # Force-reload so ASV picks up the source tree version. import cleveragents # noqa: E402 importlib.reload(cleveragents) from cleveragents.application.services.async_worker import ( # noqa: E402 AsyncWorker, AsyncWorkerConfig, CancellationToken, InMemoryJobStore, WorkerHealthReport, ) from cleveragents.domain.models.core.async_job import ( # noqa: E402 AsyncJob, AsyncJobStatus, InvalidJobTransitionError, can_transition_job, deserialize_job_payload, serialize_job_payload, ) # A valid ULID for benchmarks _ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV" _ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FAW" # --------------------------------------------------------------------------- # AsyncJob model creation benchmarks # --------------------------------------------------------------------------- class AsyncJobCreationSuite: """Benchmark AsyncJob model creation overhead.""" timeout = 60 def time_create_minimal_job(self) -> None: AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute") def time_create_job_with_payload(self) -> None: AsyncJob( job_id=_ULID_A, plan_id=_ULID_B, phase="apply", payload_json='{"key": "value", "count": 42}', ) def time_create_job_all_fields(self) -> None: AsyncJob( job_id=_ULID_A, plan_id=_ULID_B, phase="execute", status=AsyncJobStatus.QUEUED, payload_json='{"schema_version": 1}', schema_version=1, ) # --------------------------------------------------------------------------- # AsyncJob state machine benchmarks # --------------------------------------------------------------------------- class AsyncJobTransitionSuite: """Benchmark AsyncJob state machine transition overhead.""" timeout = 60 def setup(self) -> None: self._job_id = _ULID_A self._plan_id = _ULID_B def time_transition_queued_to_running(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.mark_running("worker-bench") def time_transition_running_to_succeeded(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.mark_running("worker-bench") job.mark_succeeded() def time_transition_running_to_failed(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.mark_running("worker-bench") job.mark_failed() def time_transition_queued_to_cancelled(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.mark_cancelled() def time_can_transition_check(self) -> None: can_transition_job(AsyncJobStatus.QUEUED, AsyncJobStatus.RUNNING) def time_full_lifecycle(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.mark_running("worker-bench") job.record_heartbeat() job.mark_succeeded() def time_as_cli_dict(self) -> None: job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute") job.as_cli_dict() def time_is_terminal_check(self) -> None: job = AsyncJob( job_id=self._job_id, plan_id=self._plan_id, phase="execute", status=AsyncJobStatus.QUEUED, ) _ = job.is_terminal # --------------------------------------------------------------------------- # Payload serialization benchmarks # --------------------------------------------------------------------------- class PayloadSerializationSuite: """Benchmark payload serialization/deserialization overhead.""" timeout = 60 def time_serialize_minimal(self) -> None: serialize_job_payload(_ULID_A, "execute") def time_serialize_with_extra(self) -> None: serialize_job_payload( _ULID_A, "apply", extra={"steps": ["s1", "s2"], "retries": 3} ) def time_deserialize_minimal(self) -> None: deserialize_job_payload('{"plan_id": "test", "phase": "execute"}') def time_deserialize_with_extra(self) -> None: deserialize_job_payload( '{"plan_id": "test", "phase": "apply", "extra": {"steps": ["s1"]}}' ) def time_roundtrip(self) -> None: payload = serialize_job_payload(_ULID_A, "execute", extra={"key": "val"}) deserialize_job_payload(payload) # --------------------------------------------------------------------------- # InMemoryJobStore benchmarks # --------------------------------------------------------------------------- class InMemoryJobStoreSuite: """Benchmark InMemoryJobStore CRUD operations.""" timeout = 60 def setup(self) -> None: self.store = InMemoryJobStore() # Pre-populate with 100 jobs for i in range(100): ulid = f"01ARZ3NDEKTSV4RRFFQ69G{i:04d}"[:26] job = AsyncJob(job_id=ulid, plan_id=_ULID_B, phase="execute") self.store.add(job) def time_add_job(self) -> None: store = InMemoryJobStore() job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute") store.add(job) def time_get_job(self) -> None: self.store.get(_ULID_A[:26]) def time_list_by_status(self) -> None: self.store.list_by_status(AsyncJobStatus.QUEUED) def time_list_all(self) -> None: self.store.list_all() def time_count(self) -> None: self.store.count() # --------------------------------------------------------------------------- # AsyncWorkerConfig benchmarks # --------------------------------------------------------------------------- class AsyncWorkerConfigSuite: """Benchmark AsyncWorkerConfig instantiation overhead.""" timeout = 60 def time_default_config(self) -> None: AsyncWorkerConfig() def time_custom_config(self) -> None: AsyncWorkerConfig( enabled=True, max_workers=8, poll_interval=0.5, job_timeout=7200, job_ttl=172800, ) # --------------------------------------------------------------------------- # WorkerHealthReport benchmarks # --------------------------------------------------------------------------- class WorkerHealthReportSuite: """Benchmark WorkerHealthReport lifecycle overhead.""" timeout = 60 def setup(self) -> None: self.report = WorkerHealthReport("worker-bench") def time_create_report(self) -> None: WorkerHealthReport("worker-bench") def time_record_heartbeat(self) -> None: self.report.record_heartbeat() def time_record_job_completed(self) -> None: self.report.record_job_completed() def time_record_job_failed(self) -> None: self.report.record_job_failed() def time_as_dict(self) -> None: self.report.as_dict() # --------------------------------------------------------------------------- # CancellationToken benchmarks # --------------------------------------------------------------------------- class CancellationTokenSuite: """Benchmark CancellationToken operations.""" timeout = 60 def time_create_token(self) -> None: CancellationToken() def time_check_not_cancelled(self) -> None: token = CancellationToken() _ = token.is_cancelled def time_cancel_and_check(self) -> None: token = CancellationToken() token.cancel() _ = token.is_cancelled def time_reset_token(self) -> None: token = CancellationToken() token.cancel() token.reset() def time_wait_with_timeout(self) -> None: token = CancellationToken() token.wait(timeout=0.0) # --------------------------------------------------------------------------- # AsyncWorker benchmarks # --------------------------------------------------------------------------- class AsyncWorkerSuite: """Benchmark AsyncWorker job execution overhead.""" timeout = 60 def setup(self) -> None: self.config = AsyncWorkerConfig(enabled=True) self.store = InMemoryJobStore() def time_create_worker(self) -> None: AsyncWorker(config=self.config, job_store=self.store) def time_pickup_and_execute(self) -> None: store = InMemoryJobStore() job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute") store.add(job) worker = AsyncWorker(config=self.config, job_store=store) worker.pickup_and_execute(job) def time_get_health_report(self) -> None: worker = AsyncWorker(config=self.config, job_store=self.store) worker.get_health_report() def time_detect_stuck_jobs_empty(self) -> None: worker = AsyncWorker(config=self.config, job_store=self.store) worker.detect_stuck_jobs() def time_cleanup_completed_jobs_empty(self) -> None: worker = AsyncWorker(config=self.config, job_store=self.store) worker.cleanup_completed_jobs() # --------------------------------------------------------------------------- # InvalidJobTransitionError benchmarks # --------------------------------------------------------------------------- class ErrorConstructionSuite: """Benchmark error construction overhead.""" timeout = 60 def time_invalid_transition_error(self) -> None: InvalidJobTransitionError(AsyncJobStatus.SUCCEEDED, AsyncJobStatus.RUNNING)