svcforge/tests/integration/test_tasks.py

"""TaskRepo: enqueue, complete, fail, backoff, leases."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

from svcforge_core.domain.models import TaskKind, TaskState
from svcforge_core.domain.states import InstanceState
from svcforge_core.repo.db import DictPool
from svcforge_core.repo.instances import InstanceRepo
from svcforge_core.repo.tasks import TaskRepo
from tests.integration.helpers import make_instance


async def _task_row(pool: DictPool, task_id: int) -> dict[str, object]:
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("select * from tasks where id = %s", (task_id,))
        row = await cur.fetchone()
    assert row is not None
    return dict(row)


async def test_enqueue_defaults_to_runnable_now(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    row = await _task_row(pool, tid)
    assert row["state"] == "queued"
    assert row["attempts"] == 0


async def test_complete_marks_done_and_releases_lock(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    claimed = await repo.claim("w1")
    assert claimed is not None
    await repo.complete(claimed.id, "w1")
    row = await _task_row(pool, tid)
    assert row["state"] == "done"
    assert row["locked_by"] is None


async def test_fail_under_max_attempts_requeues_with_future_run_after(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    claimed = await repo.claim("w1")
    assert claimed is not None
    assert claimed.attempts == 1  # attempts increments at CLAIM time, not on failure

    await repo.fail(tid, "helm exploded", "w1", max_attempts=5)
    row = await _task_row(pool, tid)
    assert row["state"] == "queued"
    assert row["last_error"] == "helm exploded"
    assert row["locked_by"] is None
    # Backoff pushed it out; it must not be immediately runnable again.
    run_after = row["run_after"]
    assert isinstance(run_after, datetime)
    assert run_after >= datetime.now(UTC) - timedelta(seconds=1)


async def test_fail_at_max_attempts_dead_letters_and_marks_instance(pool: DictPool) -> None:
    """A dead-letter state, not an infinite retry."""
    iid = await make_instance(pool)
    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
    tid = await tasks.enqueue_standalone(iid, TaskKind.PROVISION)

    claimed = await tasks.claim("w1")
    assert claimed is not None
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update tasks set attempts = 5 where id = %s", (tid,))

    assert await tasks.fail(tid, "chart not found", "w1", max_attempts=5) is True

    row = await _task_row(pool, tid)
    assert row["state"] == "failed"
    inst = await instances.get(iid, team="platform")
    assert inst is not None
    assert inst.state is InstanceState.FAILED
    assert inst.error == "chart not found"


async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> None:
    """Raw SQL must obey the same state machine domain.transition() enforces.

    LEGAL[DELETED] is empty — deleted is terminal. A deprovision task that exhausts its
    retries after the instance is already gone must record nothing on it, not drag it
    back to 'failed'.
    """
    iid = await make_instance(pool, state=InstanceState.DELETED)
    tasks, instances = TaskRepo(pool), InstanceRepo(pool)
    tid = await tasks.enqueue_standalone(iid, TaskKind.DEPROVISION)

    claimed = await tasks.claim("w1")
    assert claimed is not None
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update tasks set attempts = 5 where id = %s", (tid,))

    assert await tasks.fail(tid, "helm uninstall kept failing", "w1", max_attempts=5) is True

    # The task still dead-letters — that part is unconditional.
    row = await _task_row(pool, tid)
    assert row["state"] == "failed"

    # But the instance stays deleted.
    inst = await instances.get(iid, team="platform")
    assert inst is not None
    assert inst.state is InstanceState.DELETED, "raw SQL bypassed the state machine"
    assert inst.error is None


async def test_fail_truncates_error_to_2kb(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    await repo.claim("w1")
    await repo.fail(tid, "x" * 9000, "w1", max_attempts=5)
    row = await _task_row(pool, tid)
    assert isinstance(row["last_error"], str)
    assert len(row["last_error"]) == 2000


async def test_run_after_in_the_future_is_not_claimable(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    future = datetime.now(UTC) + timedelta(hours=1)
    await repo.enqueue_standalone(iid, TaskKind.PROVISION, run_after=future)
    assert await repo.claim("w1") is None


async def test_reset_expired_leases_recovers_a_dead_workers_task(pool: DictPool) -> None:
    """No distributed lock survives a power cut. Only the lease recovers this row."""
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    claimed = await repo.claim("worker-that-will-die")
    assert claimed is not None

    # Simulate: the worker was SIGKILLed 10 minutes ago and never reported.
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update tasks set locked_at = now() - interval '10 minutes' where id = %s", (tid,))

    n = await repo.reset_expired_leases(lease_seconds=300)
    assert n == 1
    row = await _task_row(pool, tid)
    assert row["state"] == TaskState.QUEUED.value
    assert row["locked_by"] is None

    # And it is claimable again.
    assert await repo.claim("w2") is not None


async def test_fresh_lease_is_not_reset(pool: DictPool) -> None:
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    await repo.enqueue_standalone(iid, TaskKind.PROVISION)
    await repo.claim("w1")
    assert await repo.reset_expired_leases(lease_seconds=300) == 0


async def test_stale_worker_cannot_complete_a_task_another_worker_now_owns(pool: DictPool) -> None:
    """The lost-lease race, as a regression test.

    Worker A hangs past its lease. The reconciler requeues the task. Worker B claims it and
    starts working. Worker A finally returns and reports success. Without the ownership
    check in `complete`, A marks the task done while B is still running it: B's helm
    install is unaccounted for, and if B then fails, the task is resurrected and a THIRD
    worker provisions the same instance. That is the double-provision the claim query's
    whole design exists to prevent, arriving through the back door.
    """
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)

    a = await repo.claim("worker-A")
    assert a is not None

    # A's lease expires and the reconciler hands the task back to the queue.
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update tasks set locked_at = now() - interval '10 minutes' where id=%s", (tid,))
    assert await repo.reset_expired_leases(lease_seconds=300) == 1

    b = await repo.claim("worker-B")
    assert b is not None, "worker-B should have been able to claim the requeued task"

    # A reports. It must lose.
    assert await repo.complete(tid, "worker-A") is False, "stale worker completed a task it no longer owns"

    row = await _task_row(pool, tid)
    assert row["state"] == "running", "the task must still belong to worker-B"
    assert row["locked_by"] == "worker-B"


async def test_stale_worker_cannot_fail_a_task_another_worker_now_owns(pool: DictPool) -> None:
    """The mirror case: a stale failure report must not requeue someone else's task."""
    iid = await make_instance(pool)
    repo = TaskRepo(pool)
    tid = await repo.enqueue_standalone(iid, TaskKind.PROVISION)

    assert await repo.claim("worker-A") is not None
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update tasks set locked_at = now() - interval '10 minutes' where id=%s", (tid,))
    await repo.reset_expired_leases(lease_seconds=300)
    assert await repo.claim("worker-B") is not None

    assert await repo.fail(tid, "stale report", "worker-A", max_attempts=5) is False

    row = await _task_row(pool, tid)
    assert row["state"] == "running"
    assert row["locked_by"] == "worker-B"
    assert row["last_error"] is None, "a stale worker wrote its error onto another worker's task"