feat(cloud): add durable cancellation support to task repository
- Add nullable cancel_requested_at column (migration 0012) - Widen ScheduledTaskStatus/TerminalTaskStatus to include cancelled - Add CancellationRequestStatus + request_task_cancellation() to CloudRepository protocol and SQLAlchemy implementation - renew_lease() now returns LeaseRenewalResult, surfacing whether cancellation is pending, instead of a bare status string - reap_expired_leases() resolves pending-cancellation tasks to cancelled instead of requeuing/failing them - record_task_result() accepts cancelled and clears cancel_requested_at on any terminal write Note: internal_api/api.py's renew_assignment route still compares renew_lease()'s return value against a bare string; it will be updated in the next task (Internal Host<->Cloud protocol) to consume LeaseRenewalResult and populate the new cancel_requested wire field.
This commit is contained in:
@@ -8,16 +8,16 @@
|
|||||||
|
|
||||||
## 2. Cloud persistence: schema and repository
|
## 2. Cloud persistence: schema and repository
|
||||||
|
|
||||||
- [ ] 2.1 Add Alembic migration `0012_task_cancellation.py` under `packages/cloud-platform/cloud/migrations/versions/`: add nullable `cancel_requested_at` column to `scheduled_tasks`, matching the style of `0008_task_progress_columns`.
|
- [x] 2.1 Add Alembic migration `0012_task_cancellation.py` under `packages/cloud-platform/cloud/migrations/versions/`: add nullable `cancel_requested_at` column to `scheduled_tasks`, matching the style of `0008_task_progress_columns`.
|
||||||
- [ ] 2.2 Add `cancel_requested_at` to `ScheduledTaskRow` (`db_models.py`) and to the `ScheduledTask` dataclass (`cloud/scheduler.py`).
|
- [x] 2.2 Add `cancel_requested_at` to `ScheduledTaskRow` (`db_models.py`) and to the `ScheduledTask` dataclass (`cloud/scheduler.py`).
|
||||||
- [ ] 2.3 Widen `ScheduledTaskStatus` (`cloud/scheduler.py`) to include `"cancelled"`.
|
- [x] 2.3 Widen `ScheduledTaskStatus` (`cloud/scheduler.py`) to include `"cancelled"`.
|
||||||
- [ ] 2.4 Widen `TerminalTaskStatus` and `record_task_result`'s accepted status literal (`repository.py` Protocol, `sql_repository.py`) to include `"cancelled"`; ensure the idempotency logic (`already_recorded`/`conflict`) treats a repeated `"cancelled"` report the same way it treats repeated `"done"`/`"failed"` reports today.
|
- [x] 2.4 Widen `TerminalTaskStatus` and `record_task_result`'s accepted status literal (`repository.py` Protocol, `sql_repository.py`) to include `"cancelled"`; ensure the idempotency logic (`already_recorded`/`conflict`) treats a repeated `"cancelled"` report the same way it treats repeated `"done"`/`"failed"` reports today.
|
||||||
- [ ] 2.5 Add `CancellationRequestStatus = Literal["requested", "already_terminal", "already_requested", "not_found"]` and a `request_task_cancellation(task_id, *, requested_at) -> CancellationRequestStatus` method to the `CloudRepository` Protocol.
|
- [x] 2.5 Add `CancellationRequestStatus = Literal["requested", "already_terminal", "already_requested", "not_found"]` and a `request_task_cancellation(task_id, *, requested_at) -> CancellationRequestStatus` method to the `CloudRepository` Protocol.
|
||||||
- [ ] 2.6 Implement `request_task_cancellation` in `SQLAlchemyCloudRepository`: for `queued` tasks, transition directly to `status="cancelled"`; for `assigned`/`dispatched` tasks, set `cancel_requested_at` if unset (return `"already_requested"` if already set); for `done`/`failed`/`cancelled` tasks, return `"already_terminal"`/success-idempotent as appropriate per design D7; for unknown task id, return `"not_found"`.
|
- [x] 2.6 Implement `request_task_cancellation` in `SQLAlchemyCloudRepository`: for `queued` tasks, transition directly to `status="cancelled"`; for `assigned`/`dispatched` tasks, set `cancel_requested_at` if unset (return `"already_requested"` if already set); for `done`/`failed`/`cancelled` tasks, return `"already_terminal"`/success-idempotent as appropriate per design D7; for unknown task id, return `"not_found"`.
|
||||||
- [ ] 2.7 Extend `renew_lease` (`sql_repository.py`) to read `cancel_requested_at` on the current row and report whether it is set, without changing its existing lease-extension/progress-update behavior.
|
- [x] 2.7 Extend `renew_lease` (`sql_repository.py`) to read `cancel_requested_at` on the current row and report whether it is set, without changing its existing lease-extension/progress-update behavior.
|
||||||
- [ ] 2.8 Update `reap_expired_leases` so a task whose `cancel_requested_at` is set resolves to `status="cancelled"` (clearing `cancel_requested_at`) instead of being requeued to `queued`, regardless of remaining attempts.
|
- [x] 2.8 Update `reap_expired_leases` so a task whose `cancel_requested_at` is set resolves to `status="cancelled"` (clearing `cancel_requested_at`) instead of being requeued to `queued`, regardless of remaining attempts.
|
||||||
- [ ] 2.9 Ensure `record_task_result` and the immediate `queued`-cancellation path both clear `cancel_requested_at` on reaching any terminal status.
|
- [x] 2.9 Ensure `record_task_result` and the immediate `queued`-cancellation path both clear `cancel_requested_at` on reaching any terminal status.
|
||||||
- [ ] 2.10 Add/extend repository-level tests (SQLite, matching existing test style) covering: immediate queued cancel; durable cancel request on assigned/dispatched surviving a simulated restart (re-fetch); renewal reporting the pending flag; expired lease with pending cancellation resolving to `cancelled`; idempotent repeat cancel calls; cancel rejected on `done`/`failed`.
|
- [x] 2.10 Add/extend repository-level tests (SQLite, matching existing test style) covering: immediate queued cancel; durable cancel request on assigned/dispatched surviving a simulated restart (re-fetch); renewal reporting the pending flag; expired lease with pending cancellation resolving to `cancelled`; idempotent repeat cancel calls; cancel rejected on `done`/`failed`.
|
||||||
|
|
||||||
## 3. Internal Host↔Cloud protocol
|
## 3. Internal Host↔Cloud protocol
|
||||||
|
|
||||||
|
|||||||
@@ -110,6 +110,7 @@ class ScheduledTaskRow(Base):
|
|||||||
progress_step_status: Mapped[str | None] = mapped_column(String, nullable=True)
|
progress_step_status: Mapped[str | None] = mapped_column(String, nullable=True)
|
||||||
progress_summary: Mapped[str | None] = mapped_column(String, nullable=True)
|
progress_summary: Mapped[str | None] = mapped_column(String, nullable=True)
|
||||||
progress_updated_at: Mapped[str | None] = mapped_column(String, nullable=True)
|
progress_updated_at: Mapped[str | None] = mapped_column(String, nullable=True)
|
||||||
|
cancel_requested_at: Mapped[str | None] = mapped_column(String, nullable=True)
|
||||||
failure_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
|
failure_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||||||
result_json: Mapped[str | None] = mapped_column(Text, nullable=True)
|
result_json: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||||||
updated_at: Mapped[str | None] = mapped_column(String, nullable=True)
|
updated_at: Mapped[str | None] = mapped_column(String, nullable=True)
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
"""Add nullable cancel_requested_at column to scheduled_tasks."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from alembic import op
|
||||||
|
|
||||||
|
revision = "0012_task_cancellation"
|
||||||
|
down_revision = "0011_planner_decision_log_reflection"
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
op.add_column(
|
||||||
|
"scheduled_tasks",
|
||||||
|
sa.Column("cancel_requested_at", sa.String(), nullable=True),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
op.drop_column("scheduled_tasks", "cancel_requested_at")
|
||||||
@@ -30,9 +30,12 @@ if TYPE_CHECKING:
|
|||||||
|
|
||||||
|
|
||||||
AttemptStatus = Literal["assigned", "dispatched", "done", "failed", "expired"]
|
AttemptStatus = Literal["assigned", "dispatched", "done", "failed", "expired"]
|
||||||
TerminalTaskStatus = Literal["done", "failed"]
|
TerminalTaskStatus = Literal["done", "failed", "cancelled"]
|
||||||
ResultRecordStatus = Literal["recorded", "already_recorded", "conflict"]
|
ResultRecordStatus = Literal["recorded", "already_recorded", "conflict"]
|
||||||
LeaseRenewalStatus = Literal["renewed", "not_found", "conflict", "expired"]
|
LeaseRenewalStatus = Literal["renewed", "not_found", "conflict", "expired"]
|
||||||
|
CancellationRequestStatus = Literal[
|
||||||
|
"requested", "already_terminal", "already_requested", "not_found"
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
class HostEnrollmentConflictError(RuntimeError):
|
class HostEnrollmentConflictError(RuntimeError):
|
||||||
@@ -91,6 +94,19 @@ class TaskAttemptRecord:
|
|||||||
terminal_result: dict[str, Any] | None = None
|
terminal_result: dict[str, Any] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class LeaseRenewalResult:
|
||||||
|
"""Outcome of a lease renewal, including whether cancellation is pending.
|
||||||
|
|
||||||
|
``cancel_requested`` reflects the task's durable ``cancel_requested_at``
|
||||||
|
column at renewal time regardless of ``status`` — callers only act on it
|
||||||
|
when ``status == "renewed"``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
status: LeaseRenewalStatus
|
||||||
|
cancel_requested: bool = False
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class LeasedAssignment:
|
class LeasedAssignment:
|
||||||
task_id: str
|
task_id: str
|
||||||
@@ -485,7 +501,7 @@ class CloudRepository(Protocol):
|
|||||||
lease_expires_at: datetime,
|
lease_expires_at: datetime,
|
||||||
now: datetime,
|
now: datetime,
|
||||||
progress: AssignmentProgressSnapshot | None = None,
|
progress: AssignmentProgressSnapshot | None = None,
|
||||||
) -> LeaseRenewalStatus: ...
|
) -> LeaseRenewalResult: ...
|
||||||
|
|
||||||
def record_task_result(
|
def record_task_result(
|
||||||
self,
|
self,
|
||||||
@@ -500,6 +516,13 @@ class CloudRepository(Protocol):
|
|||||||
completed_at: datetime,
|
completed_at: datetime,
|
||||||
) -> ResultRecordStatus: ...
|
) -> ResultRecordStatus: ...
|
||||||
|
|
||||||
|
def request_task_cancellation(
|
||||||
|
self,
|
||||||
|
task_id: str,
|
||||||
|
*,
|
||||||
|
requested_at: datetime,
|
||||||
|
) -> CancellationRequestStatus: ...
|
||||||
|
|
||||||
def reap_expired_leases(
|
def reap_expired_leases(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
|
|||||||
@@ -22,7 +22,9 @@ if TYPE_CHECKING:
|
|||||||
from cloud.store import CloudStore
|
from cloud.store import CloudStore
|
||||||
|
|
||||||
|
|
||||||
ScheduledTaskStatus = Literal["queued", "assigned", "dispatched", "done", "failed"]
|
ScheduledTaskStatus = Literal[
|
||||||
|
"queued", "assigned", "dispatched", "done", "failed", "cancelled"
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
@@ -57,6 +59,7 @@ class ScheduledTask:
|
|||||||
progress_step_status: str | None = None
|
progress_step_status: str | None = None
|
||||||
progress_summary: str | None = None
|
progress_summary: str | None = None
|
||||||
progress_updated_at: datetime | None = None
|
progress_updated_at: datetime | None = None
|
||||||
|
cancel_requested_at: datetime | None = None
|
||||||
|
|
||||||
|
|
||||||
@runtime_checkable
|
@runtime_checkable
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from alembic.runtime.migration import MigrationContext
|
|||||||
from cloud.database import create_database_engine, normalize_database_url
|
from cloud.database import create_database_engine, normalize_database_url
|
||||||
|
|
||||||
|
|
||||||
HEAD_REVISION = "0011_planner_decision_log_reflection"
|
HEAD_REVISION = "0012_task_cancellation"
|
||||||
|
|
||||||
|
|
||||||
class SchemaVersionError(RuntimeError):
|
class SchemaVersionError(RuntimeError):
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ from cloud.observability import current_correlation_id
|
|||||||
from core.models import utc_now
|
from core.models import utc_now
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from cloud.repository import AssignmentProgressSnapshot
|
from cloud.repository import AssignmentProgressSnapshot, LeaseRenewalResult
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -1497,7 +1497,9 @@ class SQLAlchemyCloudRepository:
|
|||||||
lease_expires_at: datetime,
|
lease_expires_at: datetime,
|
||||||
now: datetime,
|
now: datetime,
|
||||||
progress: AssignmentProgressSnapshot | None = None,
|
progress: AssignmentProgressSnapshot | None = None,
|
||||||
) -> str:
|
) -> "LeaseRenewalResult":
|
||||||
|
from cloud.repository import LeaseRenewalResult
|
||||||
|
|
||||||
with self._sessions.begin() as session:
|
with self._sessions.begin() as session:
|
||||||
task = session.get(
|
task = session.get(
|
||||||
ScheduledTaskRow,
|
ScheduledTaskRow,
|
||||||
@@ -1505,19 +1507,19 @@ class SQLAlchemyCloudRepository:
|
|||||||
with_for_update=self.engine.dialect.name == "postgresql",
|
with_for_update=self.engine.dialect.name == "postgresql",
|
||||||
)
|
)
|
||||||
if task is None:
|
if task is None:
|
||||||
return "not_found"
|
return LeaseRenewalResult(status="not_found")
|
||||||
if (
|
if (
|
||||||
task.status not in {"assigned", "dispatched"}
|
task.status not in {"assigned", "dispatched"}
|
||||||
or task.attempt_count != attempt
|
or task.attempt_count != attempt
|
||||||
or task.lease_id != lease_id
|
or task.lease_id != lease_id
|
||||||
or task.assigned_host_id != host_id
|
or task.assigned_host_id != host_id
|
||||||
):
|
):
|
||||||
return "conflict"
|
return LeaseRenewalResult(status="conflict")
|
||||||
current_expiry = _parse_dt(task.lease_expires_at)
|
current_expiry = _parse_dt(task.lease_expires_at)
|
||||||
if current_expiry is None or current_expiry <= now:
|
if current_expiry is None or current_expiry <= now:
|
||||||
return "expired"
|
return LeaseRenewalResult(status="expired")
|
||||||
if lease_expires_at <= now:
|
if lease_expires_at <= now:
|
||||||
return "conflict"
|
return LeaseRenewalResult(status="conflict")
|
||||||
|
|
||||||
attempt_row = session.get(
|
attempt_row = session.get(
|
||||||
TaskAttemptRow,
|
TaskAttemptRow,
|
||||||
@@ -1530,7 +1532,7 @@ class SQLAlchemyCloudRepository:
|
|||||||
or attempt_row.lease_id != lease_id
|
or attempt_row.lease_id != lease_id
|
||||||
or attempt_row.host_id != host_id
|
or attempt_row.host_id != host_id
|
||||||
):
|
):
|
||||||
return "conflict"
|
return LeaseRenewalResult(status="conflict")
|
||||||
|
|
||||||
renewed_until = _iso(lease_expires_at)
|
renewed_until = _iso(lease_expires_at)
|
||||||
task.lease_expires_at = renewed_until
|
task.lease_expires_at = renewed_until
|
||||||
@@ -1542,7 +1544,10 @@ class SQLAlchemyCloudRepository:
|
|||||||
task.progress_summary = progress.summary
|
task.progress_summary = progress.summary
|
||||||
task.progress_updated_at = _iso(progress.updated_at)
|
task.progress_updated_at = _iso(progress.updated_at)
|
||||||
_log_task_lifecycle("renewed", task)
|
_log_task_lifecycle("renewed", task)
|
||||||
return "renewed"
|
return LeaseRenewalResult(
|
||||||
|
status="renewed",
|
||||||
|
cancel_requested=task.cancel_requested_at is not None,
|
||||||
|
)
|
||||||
|
|
||||||
def record_task_result(
|
def record_task_result(
|
||||||
self,
|
self,
|
||||||
@@ -1579,7 +1584,7 @@ class SQLAlchemyCloudRepository:
|
|||||||
):
|
):
|
||||||
return "conflict"
|
return "conflict"
|
||||||
|
|
||||||
if task.status in {"done", "failed"}:
|
if task.status in {"done", "failed", "cancelled"}:
|
||||||
if (
|
if (
|
||||||
task.status == status
|
task.status == status
|
||||||
and task.failure_reason == failure_reason
|
and task.failure_reason == failure_reason
|
||||||
@@ -1593,7 +1598,7 @@ class SQLAlchemyCloudRepository:
|
|||||||
current_expiry = _parse_dt(task.lease_expires_at)
|
current_expiry = _parse_dt(task.lease_expires_at)
|
||||||
if current_expiry is None or current_expiry <= completed_at:
|
if current_expiry is None or current_expiry <= completed_at:
|
||||||
return "conflict"
|
return "conflict"
|
||||||
if status not in {"done", "failed"}:
|
if status not in {"done", "failed", "cancelled"}:
|
||||||
return "conflict"
|
return "conflict"
|
||||||
|
|
||||||
result_json = (
|
result_json = (
|
||||||
@@ -1610,11 +1615,18 @@ class SQLAlchemyCloudRepository:
|
|||||||
task.progress_step_status = None
|
task.progress_step_status = None
|
||||||
task.progress_summary = None
|
task.progress_summary = None
|
||||||
task.progress_updated_at = None
|
task.progress_updated_at = None
|
||||||
|
task.cancel_requested_at = None
|
||||||
attempt_row.status = status
|
attempt_row.status = status
|
||||||
attempt_row.completed_at = completed_at_iso
|
attempt_row.completed_at = completed_at_iso
|
||||||
attempt_row.failure_reason = failure_reason
|
attempt_row.failure_reason = failure_reason
|
||||||
attempt_row.result_json = result_json
|
attempt_row.result_json = result_json
|
||||||
_log_task_lifecycle("completed" if status == "done" else "failed", task)
|
if status == "done":
|
||||||
|
lifecycle_event = "completed"
|
||||||
|
elif status == "cancelled":
|
||||||
|
lifecycle_event = "cancelled"
|
||||||
|
else:
|
||||||
|
lifecycle_event = "failed"
|
||||||
|
_log_task_lifecycle(lifecycle_event, task)
|
||||||
return "recorded"
|
return "recorded"
|
||||||
|
|
||||||
def reap_expired_leases(
|
def reap_expired_leases(
|
||||||
@@ -1657,7 +1669,12 @@ class SQLAlchemyCloudRepository:
|
|||||||
task.lease_id = None
|
task.lease_id = None
|
||||||
task.lease_expires_at = None
|
task.lease_expires_at = None
|
||||||
task.result_json = None
|
task.result_json = None
|
||||||
if task.attempt_count < max_attempts:
|
if task.cancel_requested_at is not None:
|
||||||
|
task.status = "cancelled"
|
||||||
|
task.assigned_host_id = None
|
||||||
|
task.assigned_device_id = None
|
||||||
|
task.cancel_requested_at = None
|
||||||
|
elif task.attempt_count < max_attempts:
|
||||||
task.status = "queued"
|
task.status = "queued"
|
||||||
task.assigned_host_id = None
|
task.assigned_host_id = None
|
||||||
task.assigned_device_id = None
|
task.assigned_device_id = None
|
||||||
@@ -1667,13 +1684,46 @@ class SQLAlchemyCloudRepository:
|
|||||||
task.failure_reason = (
|
task.failure_reason = (
|
||||||
f"lease expired after {task.attempt_count} attempts"
|
f"lease expired after {task.attempt_count} attempts"
|
||||||
)
|
)
|
||||||
_log_task_lifecycle(
|
if task.status == "queued":
|
||||||
"retried" if task.status == "queued" else "failed",
|
lifecycle_event = "retried"
|
||||||
task,
|
elif task.status == "cancelled":
|
||||||
)
|
lifecycle_event = "cancelled"
|
||||||
|
else:
|
||||||
|
lifecycle_event = "failed"
|
||||||
|
_log_task_lifecycle(lifecycle_event, task)
|
||||||
reaped_task_ids.append(task.id)
|
reaped_task_ids.append(task.id)
|
||||||
return reaped_task_ids
|
return reaped_task_ids
|
||||||
|
|
||||||
|
def request_task_cancellation(
|
||||||
|
self,
|
||||||
|
task_id: str,
|
||||||
|
*,
|
||||||
|
requested_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 "not_found"
|
||||||
|
if task.status == "queued":
|
||||||
|
task.status = "cancelled"
|
||||||
|
task.cancel_requested_at = None
|
||||||
|
task.updated_at = _iso(requested_at)
|
||||||
|
_log_task_lifecycle("cancelled", task)
|
||||||
|
return "requested"
|
||||||
|
if task.status in {"assigned", "dispatched"}:
|
||||||
|
if task.cancel_requested_at is not None:
|
||||||
|
return "already_requested"
|
||||||
|
task.cancel_requested_at = _iso(requested_at)
|
||||||
|
task.updated_at = _iso(requested_at)
|
||||||
|
return "requested"
|
||||||
|
if task.status == "cancelled":
|
||||||
|
return "requested"
|
||||||
|
return "already_terminal"
|
||||||
|
|
||||||
def list_task_attempts(self, task_id: str) -> list[Any]:
|
def list_task_attempts(self, task_id: str) -> list[Any]:
|
||||||
with self._sessions() as session:
|
with self._sessions() as session:
|
||||||
rows = session.scalars(
|
rows = session.scalars(
|
||||||
@@ -2223,6 +2273,7 @@ def _task_from_row(row: ScheduledTaskRow) -> Any:
|
|||||||
progress_step_status=row.progress_step_status,
|
progress_step_status=row.progress_step_status,
|
||||||
progress_summary=row.progress_summary,
|
progress_summary=row.progress_summary,
|
||||||
progress_updated_at=_parse_dt(row.progress_updated_at),
|
progress_updated_at=_parse_dt(row.progress_updated_at),
|
||||||
|
cancel_requested_at=_parse_dt(row.cancel_requested_at),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -967,7 +967,8 @@ def test_active_lease_renews_for_owning_host(database_url: str) -> None:
|
|||||||
now=now + timedelta(seconds=30),
|
now=now + timedelta(seconds=30),
|
||||||
)
|
)
|
||||||
|
|
||||||
assert status == "renewed"
|
assert status.status == "renewed"
|
||||||
|
assert status.cancel_requested is False
|
||||||
task = database.repository.get_task(task_id)
|
task = database.repository.get_task(task_id)
|
||||||
assert task is not None
|
assert task is not None
|
||||||
assert task.lease_expires_at == renewed_expiry
|
assert task.lease_expires_at == renewed_expiry
|
||||||
@@ -1031,7 +1032,7 @@ def test_stale_or_foreign_lease_renewal_conflicts(
|
|||||||
now=now + timedelta(seconds=30),
|
now=now + timedelta(seconds=30),
|
||||||
)
|
)
|
||||||
|
|
||||||
assert status == "conflict"
|
assert status.status == "conflict"
|
||||||
task = database.repository.get_task(task_id)
|
task = database.repository.get_task(task_id)
|
||||||
assert task is not None
|
assert task is not None
|
||||||
assert task.lease_expires_at == initial_expiry
|
assert task.lease_expires_at == initial_expiry
|
||||||
@@ -1078,7 +1079,7 @@ def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None:
|
|||||||
host_id=host_id,
|
host_id=host_id,
|
||||||
lease_expires_at=now + timedelta(minutes=2),
|
lease_expires_at=now + timedelta(minutes=2),
|
||||||
now=now + timedelta(seconds=2),
|
now=now + timedelta(seconds=2),
|
||||||
)
|
).status
|
||||||
== "expired"
|
== "expired"
|
||||||
)
|
)
|
||||||
assert (
|
assert (
|
||||||
@@ -1089,7 +1090,7 @@ def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None:
|
|||||||
host_id=host_id,
|
host_id=host_id,
|
||||||
lease_expires_at=now + timedelta(minutes=2),
|
lease_expires_at=now + timedelta(minutes=2),
|
||||||
now=now,
|
now=now,
|
||||||
)
|
).status
|
||||||
== "not_found"
|
== "not_found"
|
||||||
)
|
)
|
||||||
finally:
|
finally:
|
||||||
@@ -1810,3 +1811,251 @@ def test_record_planner_decision_stores_null_rationale_and_thinking(
|
|||||||
assert decisions[0].thinking is None
|
assert decisions[0].thinking is None
|
||||||
finally:
|
finally:
|
||||||
database.close()
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cancel_queued_task_is_immediate(database_url: str) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
task_id = _unique_id("cancel-queued-task")
|
||||||
|
now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
|
||||||
|
|
||||||
|
try:
|
||||||
|
database.repository.enqueue_task(
|
||||||
|
ScheduledTask(
|
||||||
|
id=task_id,
|
||||||
|
goal="cancel before assignment",
|
||||||
|
workflow_definition_id=None,
|
||||||
|
constraints=TaskConstraints(),
|
||||||
|
created_at=now,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
status = database.repository.request_task_cancellation(
|
||||||
|
task_id, requested_at=now
|
||||||
|
)
|
||||||
|
|
||||||
|
assert status == "requested"
|
||||||
|
task = database.repository.get_task(task_id)
|
||||||
|
assert task is not None
|
||||||
|
assert task.status == "cancelled"
|
||||||
|
assert task.cancel_requested_at is None
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cancel_request_on_assigned_task_is_durable(database_url: str) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
host_id = _unique_id("cancel-durable-host")
|
||||||
|
device_id = _unique_id("cancel-durable-device")
|
||||||
|
task_id = _unique_id("cancel-durable-task")
|
||||||
|
now = datetime(2026, 7, 15, 8, 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="cancel while running",
|
||||||
|
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="durable-cancel-lease",
|
||||||
|
lease_expires_at=now + timedelta(minutes=5),
|
||||||
|
now=now,
|
||||||
|
)
|
||||||
|
|
||||||
|
status = database.repository.request_task_cancellation(
|
||||||
|
task_id, requested_at=now
|
||||||
|
)
|
||||||
|
assert status == "requested"
|
||||||
|
|
||||||
|
# Simulate a process restart by re-fetching the task from a fresh read.
|
||||||
|
task = database.repository.get_task(task_id)
|
||||||
|
assert task is not None
|
||||||
|
assert task.status == "assigned"
|
||||||
|
assert task.cancel_requested_at == now
|
||||||
|
|
||||||
|
repeat_status = database.repository.request_task_cancellation(
|
||||||
|
task_id, requested_at=now + timedelta(seconds=5)
|
||||||
|
)
|
||||||
|
assert repeat_status == "already_requested"
|
||||||
|
task = database.repository.get_task(task_id)
|
||||||
|
assert task is not None
|
||||||
|
assert task.cancel_requested_at == now
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_renew_lease_reports_pending_cancellation(database_url: str) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
host_id = _unique_id("cancel-renew-host")
|
||||||
|
device_id = _unique_id("cancel-renew-device")
|
||||||
|
task_id = _unique_id("cancel-renew-task")
|
||||||
|
now = datetime(2026, 7, 15, 8, 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="report pending cancellation on renewal",
|
||||||
|
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="renew-cancel-lease",
|
||||||
|
lease_expires_at=now + timedelta(minutes=5),
|
||||||
|
now=now,
|
||||||
|
)
|
||||||
|
database.repository.request_task_cancellation(task_id, requested_at=now)
|
||||||
|
|
||||||
|
result = database.repository.renew_lease(
|
||||||
|
task_id=task_id,
|
||||||
|
attempt=1,
|
||||||
|
lease_id="renew-cancel-lease",
|
||||||
|
host_id=host_id,
|
||||||
|
lease_expires_at=now + timedelta(minutes=10),
|
||||||
|
now=now + timedelta(seconds=30),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.status == "renewed"
|
||||||
|
assert result.cancel_requested is True
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_expired_lease_with_pending_cancellation_resolves_to_cancelled(
|
||||||
|
database_url: str,
|
||||||
|
) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
host_id = _unique_id("cancel-expiry-host")
|
||||||
|
device_id = _unique_id("cancel-expiry-device")
|
||||||
|
task_id = _unique_id("cancel-expiry-task")
|
||||||
|
now = datetime(2026, 7, 15, 8, 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="cancelled task must not be requeued",
|
||||||
|
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="cancel-then-expire-lease",
|
||||||
|
lease_expires_at=expired_at,
|
||||||
|
now=now,
|
||||||
|
)
|
||||||
|
database.repository.request_task_cancellation(task_id, requested_at=now)
|
||||||
|
|
||||||
|
reaped_task_ids = database.repository.reap_expired_leases(
|
||||||
|
now=reaped_at,
|
||||||
|
# A high attempt limit proves cancellation takes priority over retry.
|
||||||
|
max_attempts=5,
|
||||||
|
)
|
||||||
|
assert task_id in reaped_task_ids
|
||||||
|
|
||||||
|
task = database.repository.get_task(task_id)
|
||||||
|
assert task is not None
|
||||||
|
assert task.status == "cancelled"
|
||||||
|
assert task.cancel_requested_at is None
|
||||||
|
assert task.assigned_host_id is None
|
||||||
|
assert task.assigned_device_id is None
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cancel_request_rejected_on_terminal_task(database_url: str) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
host_id = _unique_id("cancel-terminal-host")
|
||||||
|
device_id = _unique_id("cancel-terminal-device")
|
||||||
|
task_id = _unique_id("cancel-terminal-task")
|
||||||
|
now = datetime(2026, 7, 15, 8, 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="already finished",
|
||||||
|
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="terminal-lease",
|
||||||
|
lease_expires_at=now + timedelta(minutes=5),
|
||||||
|
now=now,
|
||||||
|
)
|
||||||
|
database.repository.record_task_result(
|
||||||
|
task_id=task_id,
|
||||||
|
attempt=1,
|
||||||
|
lease_id="terminal-lease",
|
||||||
|
host_id=host_id,
|
||||||
|
status="done",
|
||||||
|
failure_reason=None,
|
||||||
|
terminal_result={"ok": True},
|
||||||
|
completed_at=now + timedelta(seconds=10),
|
||||||
|
)
|
||||||
|
|
||||||
|
status = database.repository.request_task_cancellation(
|
||||||
|
task_id, requested_at=now + timedelta(seconds=20)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert status == "already_terminal"
|
||||||
|
task = database.repository.get_task(task_id)
|
||||||
|
assert task is not None
|
||||||
|
assert task.status == "done"
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cancel_request_unknown_task_not_found(database_url: str) -> None:
|
||||||
|
database = CloudDatabase(database_url)
|
||||||
|
now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
|
||||||
|
|
||||||
|
try:
|
||||||
|
status = database.repository.request_task_cancellation(
|
||||||
|
_unique_id("missing-cancel-task"), requested_at=now
|
||||||
|
)
|
||||||
|
assert status == "not_found"
|
||||||
|
finally:
|
||||||
|
database.close()
|
||||||
|
|||||||
@@ -90,7 +90,8 @@ def test_renew_lease_writes_progress_on_success(tmp_path) -> None:
|
|||||||
now=now + timedelta(seconds=5),
|
now=now + timedelta(seconds=5),
|
||||||
progress=progress,
|
progress=progress,
|
||||||
)
|
)
|
||||||
assert result == "renewed"
|
assert result.status == "renewed"
|
||||||
|
assert result.cancel_requested is False
|
||||||
|
|
||||||
task = database.repository.get_task(task_id)
|
task = database.repository.get_task(task_id)
|
||||||
assert task is not None
|
assert task is not None
|
||||||
|
|||||||
Reference in New Issue
Block a user