diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index d73b07d..72313f2 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -23,7 +23,7 @@ - [x] 3.4 Implement lease renewal with host/task/attempt ownership validation and conflict responses for stale leases. - [x] 3.5 Implement idempotent terminal result recording and reservation release for active leases. - [x] 3.6 Implement expired-lease requeue/failure behavior with bounded attempts and preserved attempt history. -- [ ] 3.7 Add PostgreSQL concurrency tests proving one assignment/claim winner and SQLite tests documenting single-control-plane behavior. +- [x] 3.7 Add PostgreSQL concurrency tests proving one assignment/claim winner and SQLite tests documenting single-control-plane behavior. ## 4. Authentication And Authorization diff --git a/tests/test_cloud_repository_concurrency.py b/tests/test_cloud_repository_concurrency.py new file mode 100644 index 0000000..e002c93 --- /dev/null +++ b/tests/test_cloud_repository_concurrency.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import os +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta +from pathlib import Path +from threading import Barrier +from uuid import uuid4 + +import pytest + +from cloud.database import CloudDatabase +from cloud.pool import PooledDevice +from cloud.scheduler import ScheduledTask, TaskConstraints + + +def _unique_id(prefix: str) -> str: + return f"{prefix}-{uuid4().hex}" + + +def _seed_assignment_candidate( + database: CloudDatabase, + *, + now: datetime, +) -> tuple[str, str, str]: + host_id = _unique_id("concurrency-host") + device_id = _unique_id("concurrency-device") + task_id = _unique_id("concurrency-task") + database.repository.upsert_host(host_id, address=None, last_seen_at=now) + database.repository.replace_host_devices( + host_id, + [ + PooledDevice( + device_id=device_id, + host_id=host_id, + driver_type="wda", + status="idle", + synced_at=now, + ) + ], + ) + database.repository.enqueue_task( + ScheduledTask( + id=task_id, + goal="concurrent operation", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + return host_id, device_id, task_id + + +def _postgres_url() -> str: + url = os.getenv("TEST_POSTGRES_URL") + if not url: + pytest.skip("TEST_POSTGRES_URL is required for PostgreSQL concurrency tests") + return url + + +@pytest.mark.integration +def test_postgresql_concurrent_assignment_has_one_winner() -> None: + database_url = _postgres_url() + setup = CloudDatabase(database_url) + now = datetime(2026, 7, 12, 17, 0, tzinfo=UTC) + host_id, device_id, task_id = _seed_assignment_candidate(setup, now=now) + setup.close() + barrier = Barrier(2) + + def assign(lease_id: str): + contender = CloudDatabase(database_url, create_schema=False) + try: + barrier.wait(timeout=5) + return contender.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id=lease_id, + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + finally: + contender.close() + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(assign, ("lease-a", "lease-b"))) + + assert sum(result is not None for result in results) == 1 + verification = CloudDatabase(database_url, create_schema=False) + try: + task = verification.repository.get_task(task_id) + attempts = verification.repository.list_task_attempts(task_id) + assert task is not None + assert task.status == "assigned" + assert task.attempt_count == 1 + assert len(attempts) == 1 + assert attempts[0].lease_id in {"lease-a", "lease-b"} + finally: + verification.close() + + +@pytest.mark.integration +def test_postgresql_concurrent_claim_has_one_winner() -> None: + database_url = _postgres_url() + setup = CloudDatabase(database_url) + now = datetime(2026, 7, 12, 18, 0, tzinfo=UTC) + host_id, device_id, task_id = _seed_assignment_candidate(setup, now=now) + setup.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="claim-race-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + setup.close() + barrier = Barrier(2) + + def claim(): + contender = CloudDatabase(database_url, create_schema=False) + try: + barrier.wait(timeout=5) + return contender.repository.claim_assignment(host_id=host_id, now=now) + finally: + contender.close() + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(lambda _index: claim(), range(2))) + + assert sum(result is not None for result in results) == 1 + verification = CloudDatabase(database_url, create_schema=False) + try: + task = verification.repository.get_task(task_id) + attempts = verification.repository.list_task_attempts(task_id) + assert task is not None + assert task.status == "dispatched" + assert len(attempts) == 1 + assert attempts[0].status == "dispatched" + finally: + verification.close() + + +def test_sqlite_single_control_plane_rejects_sequential_duplicate_assignment( + tmp_path: Path, +) -> None: + database = CloudDatabase(f"sqlite:///{(tmp_path / 'single.sqlite3').as_posix()}") + now = datetime(2026, 7, 12, 19, 0, tzinfo=UTC) + host_id, device_id, task_id = _seed_assignment_candidate(database, now=now) + + try: + first = database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="sqlite-lease-a", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + second = database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="sqlite-lease-b", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + + assert database.engine.dialect.name == "sqlite" + assert first is not None + assert second is None + assert len(database.repository.list_task_attempts(task_id)) == 1 + finally: + database.close()