feat(cloud-scheduler): record terminal results

2026-07-12 17:25:57 +08:00
parent 1b15a8a218
commit e1af4403f5
4 changed files with 278 additions and 1 deletions
@@ -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()