diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 0dceb72..d73b07d 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -22,7 +22,7 @@ - [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. - [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. +- [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. ## 4. Authentication And Authorization diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index b6353cb..5c7218c 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/packages/cloud-platform/cloud/sql_repository.py @@ -451,6 +451,59 @@ class SQLAlchemyCloudRepository: attempt_row.result_json = result_json return "recorded" + def reap_expired_leases( + self, + *, + now: datetime, + max_attempts: int, + ) -> list[str]: + if max_attempts < 1: + raise ValueError("max_attempts must be at least 1") + + with self._sessions.begin() as session: + statement = ( + select(ScheduledTaskRow) + .where( + ScheduledTaskRow.status.in_(("assigned", "dispatched")), + ScheduledTaskRow.lease_expires_at.is_not(None), + ScheduledTaskRow.lease_expires_at <= _iso(now), + ) + .order_by(ScheduledTaskRow.lease_expires_at, ScheduledTaskRow.id) + ) + if self.engine.dialect.name == "postgresql": + statement = statement.with_for_update(skip_locked=True) + + tasks = session.scalars(statement).all() + reaped_task_ids: list[str] = [] + for task in tasks: + attempt_row = session.get( + TaskAttemptRow, + (task.id, task.attempt_count), + with_for_update=self.engine.dialect.name == "postgresql", + ) + if attempt_row is None: + continue + + attempt_row.status = "expired" + attempt_row.completed_at = _iso(now) + attempt_row.failure_reason = "lease expired" + task.updated_at = _iso(now) + task.lease_id = None + task.lease_expires_at = None + task.result_json = None + if task.attempt_count < max_attempts: + task.status = "queued" + task.assigned_host_id = None + task.assigned_device_id = None + task.failure_reason = None + else: + task.status = "failed" + task.failure_reason = ( + f"lease expired after {task.attempt_count} attempts" + ) + reaped_task_ids.append(task.id) + return reaped_task_ids + 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 412172e..3198d1c 100644 --- a/tests/test_cloud_repository_contract.py +++ b/tests/test_cloud_repository_contract.py @@ -957,3 +957,191 @@ def test_stale_foreign_or_expired_result_is_rejected( assert task.terminal_result is None finally: database.close() + + +@pytest.mark.parametrize("claimed", [False, True], ids=["assigned", "dispatched"]) +def test_expired_lease_requeues_with_auditable_history( + database_url: str, + claimed: bool, +) -> None: + database = CloudDatabase(database_url) + host_id = _unique_id("requeue-host") + device_id = _unique_id("requeue-device") + task_id = _unique_id("requeue-task") + now = datetime(2026, 7, 12, 14, 0, tzinfo=UTC) + expired_at = now + timedelta(seconds=10) + reaped_at = expired_at + timedelta(seconds=1) + + 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="retry after expiry", + 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="expired-attempt-1", + lease_expires_at=expired_at, + now=now, + ) + if claimed: + database.repository.claim_assignment(host_id=host_id, now=now) + + reaped_task_ids = database.repository.reap_expired_leases( + now=reaped_at, + max_attempts=2, + ) + assert task_id in reaped_task_ids + + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "queued" + assert task.attempt_count == 1 + assert task.assigned_host_id is None + assert task.assigned_device_id is None + assert task.lease_id is None + assert task.lease_expires_at is None + assert task.failure_reason is None + attempts = database.repository.list_task_attempts(task_id) + assert len(attempts) == 1 + assert attempts[0].status == "expired" + assert attempts[0].failure_reason == "lease expired" + assert attempts[0].completed_at == reaped_at + assert device_id not in database.repository.list_reserved_device_ids( + now=reaped_at + ) + + second_assignment = database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="active-attempt-2", + lease_expires_at=reaped_at + timedelta(minutes=1), + now=reaped_at, + ) + assert second_assignment is not None + assert second_assignment.attempt == 2 + assert [ + attempt.status + for attempt in database.repository.list_task_attempts(task_id) + ] == ["expired", "assigned"] + finally: + database.close() + + +def test_expired_lease_fails_at_attempt_limit(database_url: str) -> None: + database = CloudDatabase(database_url) + host_id = _unique_id("limit-host") + device_id = _unique_id("limit-device") + task_id = _unique_id("limit-task") + now = datetime(2026, 7, 12, 15, 0, tzinfo=UTC) + expired_at = now + timedelta(seconds=10) + reaped_at = expired_at + timedelta(seconds=1) + + 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="fail after expiry", + 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="final-attempt", + lease_expires_at=expired_at, + now=now, + ) + + reaped_task_ids = database.repository.reap_expired_leases( + now=reaped_at, + max_attempts=1, + ) + assert task_id in reaped_task_ids + + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "failed" + assert task.failure_reason == "lease expired after 1 attempts" + assert task.lease_id is None + assert task.lease_expires_at is None + attempts = database.repository.list_task_attempts(task_id) + assert attempts[0].status == "expired" + assert device_id not in database.repository.list_reserved_device_ids( + now=reaped_at + ) + assert ( + database.repository.reap_expired_leases( + now=reaped_at, + max_attempts=1, + ) + == [] + ) + finally: + database.close() + + +def test_unexpired_lease_is_not_reaped(database_url: str) -> None: + database = CloudDatabase(database_url) + host_id = _unique_id("active-reaper-host") + device_id = _unique_id("active-reaper-device") + task_id = _unique_id("active-reaper-task") + now = datetime(2026, 7, 12, 16, 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="stay active", + 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="active-reaper-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + + assert ( + database.repository.reap_expired_leases( + now=now + timedelta(seconds=30), + max_attempts=2, + ) + == [] + ) + task = database.repository.get_task(task_id) + assert task is not None + assert task.status == "assigned" + assert device_id in database.repository.list_reserved_device_ids(now=now) + finally: + database.close()