From e1af4403f5b6c3f9b0f88db4d8bbed5380bddf38 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Sun, 12 Jul 2026 17:25:57 +0800 Subject: [PATCH] feat(cloud-scheduler): record terminal results --- .../cloud-control-plane-integration/tasks.md | 2 +- packages/cloud-platform/cloud/repository.py | 1 + .../cloud-platform/cloud/sql_repository.py | 68 ++++++ tests/test_cloud_repository_contract.py | 208 ++++++++++++++++++ 4 files changed, 278 insertions(+), 1 deletion(-) diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 467066d..0dceb72 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -21,7 +21,7 @@ - [x] 3.2 Implement atomic queued-task assignment and device reservation while excluding devices with active assignments even when snapshots report idle. - [x] 3.3 Implement owning-host claim that atomically transitions one assigned attempt to dispatched under its active lease. - [x] 3.4 Implement lease renewal with host/task/attempt ownership validation and conflict responses for stale leases. -- [ ] 3.5 Implement idempotent terminal result recording and reservation release for active leases. +- [x] 3.5 Implement idempotent terminal result recording and reservation release for active leases. - [ ] 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. diff --git a/packages/cloud-platform/cloud/repository.py b/packages/cloud-platform/cloud/repository.py index 8b92889..ac08aa6 100644 --- a/packages/cloud-platform/cloud/repository.py +++ b/packages/cloud-platform/cloud/repository.py @@ -131,6 +131,7 @@ class CloudRepository(Protocol): host_id: str, status: TerminalTaskStatus, failure_reason: str | None, + terminal_result: dict[str, Any] | None, completed_at: datetime, ) -> ResultRecordStatus: ... diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index 6e866a6..b6353cb 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/packages/cloud-platform/cloud/sql_repository.py @@ -383,6 +383,74 @@ class SQLAlchemyCloudRepository: attempt_row.lease_expires_at = renewed_until return "renewed" + def record_task_result( + self, + *, + task_id: str, + attempt: int, + lease_id: str, + host_id: str, + status: str, + failure_reason: str | None, + terminal_result: dict[str, Any] | None, + completed_at: datetime, + ) -> str: + with self._sessions.begin() as session: + task = session.get( + ScheduledTaskRow, + task_id, + with_for_update=self.engine.dialect.name == "postgresql", + ) + if task is None: + return "conflict" + attempt_row = session.get( + TaskAttemptRow, + (task_id, attempt), + with_for_update=self.engine.dialect.name == "postgresql", + ) + if ( + attempt_row is None + or task.attempt_count != attempt + or task.lease_id != lease_id + or task.assigned_host_id != host_id + or attempt_row.lease_id != lease_id + or attempt_row.host_id != host_id + ): + return "conflict" + + if task.status in {"done", "failed"}: + if ( + task.status == status + and task.failure_reason == failure_reason + and _parse_json_object(task.result_json) == terminal_result + and attempt_row.status == status + ): + return "already_recorded" + return "conflict" + if task.status not in {"assigned", "dispatched"}: + return "conflict" + current_expiry = _parse_dt(task.lease_expires_at) + if current_expiry is None or current_expiry <= completed_at: + return "conflict" + if status not in {"done", "failed"}: + return "conflict" + + result_json = ( + json.dumps(terminal_result, ensure_ascii=False) + if terminal_result is not None + else None + ) + completed_at_iso = _iso(completed_at) + task.status = status + task.failure_reason = failure_reason + task.result_json = result_json + task.updated_at = completed_at_iso + attempt_row.status = status + attempt_row.completed_at = completed_at_iso + attempt_row.failure_reason = failure_reason + attempt_row.result_json = result_json + return "recorded" + def list_task_attempts(self, task_id: str) -> list[Any]: with self._sessions() as session: rows = session.scalars( diff --git a/tests/test_cloud_repository_contract.py b/tests/test_cloud_repository_contract.py index f5e8248..412172e 100644 --- a/tests/test_cloud_repository_contract.py +++ b/tests/test_cloud_repository_contract.py @@ -749,3 +749,211 @@ def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None: ) finally: database.close() + + +def test_terminal_result_is_recorded_idempotently_and_releases_reservation( + database_url: str, +) -> None: + database = CloudDatabase(database_url) + host_id = _unique_id("result-host") + device_id = _unique_id("result-device") + task_id = _unique_id("result-task") + now = datetime(2026, 7, 12, 11, 0, tzinfo=UTC) + completed_at = now + timedelta(seconds=30) + result = {"steps": 4, "summary": "completed"} + + try: + database.repository.upsert_host(host_id, address=None, last_seen_at=now) + database.repository.replace_host_devices( + host_id, + [_device(device_id, host_id)], + ) + database.repository.enqueue_task( + ScheduledTask( + id=task_id, + goal="complete me", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="result-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + database.repository.claim_assignment(host_id=host_id, now=now) + + first_status = database.repository.record_task_result( + task_id=task_id, + attempt=1, + lease_id="result-lease", + host_id=host_id, + status="done", + failure_reason=None, + terminal_result=result, + completed_at=completed_at, + ) + repeated_status = database.repository.record_task_result( + task_id=task_id, + attempt=1, + lease_id="result-lease", + host_id=host_id, + status="done", + failure_reason=None, + terminal_result=result, + completed_at=completed_at + timedelta(seconds=1), + ) + + assert first_status == "recorded" + assert repeated_status == "already_recorded" + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "done" + assert task.terminal_result == result + attempts = database.repository.list_task_attempts(task_id) + assert attempts[0].status == "done" + assert attempts[0].terminal_result == result + assert attempts[0].completed_at == completed_at + assert device_id not in database.repository.list_reserved_device_ids( + now=completed_at + ) + finally: + database.close() + + +def test_conflicting_terminal_result_cannot_overwrite_recorded_outcome( + database_url: str, +) -> None: + database = CloudDatabase(database_url) + host_id = _unique_id("result-conflict-host") + device_id = _unique_id("result-conflict-device") + task_id = _unique_id("result-conflict-task") + now = datetime(2026, 7, 12, 12, 0, tzinfo=UTC) + + try: + database.repository.upsert_host(host_id, address=None, last_seen_at=now) + database.repository.replace_host_devices( + host_id, + [_device(device_id, host_id)], + ) + database.repository.enqueue_task( + ScheduledTask( + id=task_id, + goal="stable outcome", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="stable-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + assert ( + database.repository.record_task_result( + task_id=task_id, + attempt=1, + lease_id="stable-lease", + host_id=host_id, + status="failed", + failure_reason="device offline", + terminal_result={"retryable": True}, + completed_at=now + timedelta(seconds=10), + ) + == "recorded" + ) + + assert ( + database.repository.record_task_result( + task_id=task_id, + attempt=1, + lease_id="stable-lease", + host_id=host_id, + status="done", + failure_reason=None, + terminal_result={"retryable": False}, + completed_at=now + timedelta(seconds=20), + ) + == "conflict" + ) + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "failed" + assert task.failure_reason == "device offline" + assert task.terminal_result == {"retryable": True} + finally: + database.close() + + +@pytest.mark.parametrize( + ("attempt", "lease_id", "host_kind", "report_delay"), + [ + (2, "active-lease", "owner", 10), + (1, "stale-lease", "owner", 10), + (1, "active-lease", "foreign", 10), + (1, "active-lease", "owner", 61), + ], +) +def test_stale_foreign_or_expired_result_is_rejected( + database_url: str, + attempt: int, + lease_id: str, + host_kind: str, + report_delay: int, +) -> None: + database = CloudDatabase(database_url) + owner_id = _unique_id("result-owner") + device_id = _unique_id("result-stale-device") + task_id = _unique_id("result-stale-task") + now = datetime(2026, 7, 12, 13, 0, tzinfo=UTC) + + try: + database.repository.upsert_host(owner_id, address=None, last_seen_at=now) + database.repository.replace_host_devices( + owner_id, + [_device(device_id, owner_id)], + ) + database.repository.enqueue_task( + ScheduledTask( + id=task_id, + goal="reject stale result", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + database.repository.assign_task( + task_id=task_id, + host_id=owner_id, + device_id=device_id, + lease_id="active-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + + status = database.repository.record_task_result( + task_id=task_id, + attempt=attempt, + lease_id=lease_id, + host_id=owner_id if host_kind == "owner" else _unique_id("foreign"), + status="done", + failure_reason=None, + terminal_result={"ignored": True}, + completed_at=now + timedelta(seconds=report_delay), + ) + + assert status == "conflict" + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "assigned" + assert task.terminal_result is None + finally: + database.close()