Nguyen Minh Phuc
d64c3c9f39
fail() blanket-marked the instance `failed` for every dead-lettered task kind.
Only provision is correct. The others each left the instance in a state the
UPDATE then corrupted:
- deprovision: `deleting` -> `failed` stranded the instance, because
due_for_deprovision only re-selects ready/deleting, leaking the release
reconcile.py promises to reclaim.
- upgrade: helm --atomic rolled back, so the instance was still `ready` and
serving the old version; `failed` mislabelled a healthy service and dropped
it off the upgrade work-list.
- verify: handle_verify already halted the rollout; the instance was `ready`.
Now gated on kind == provision, with a regression test per kind (control-tested
against the blanket UPDATE, which fails all three).
The API exception handler was registered on fastapi.HTTPException, a subclass
of starlette's. Starlette matches handlers by walking type(exc).__mro__, so
framework-raised 404/405 never hit it and returned {"detail": ...} instead of
ErrorBody. Registered on the starlette parent, and added a
RequestValidationError handler so body-validation 422s share the shape too.
Tests assert the ErrorBody shape for framework 404, 405, and a forbidden field.
helm._list_releases_via_api built the httpx client with verify=<ca path>, which
loads the CA eagerly and raises OSError on a half-mounted ServiceAccount — an
error the except clause did not catch, crashing the reconciler tick as a bare
bug. Now the CA is checked for readability alongside the token, and a missing
one means "not in-cluster" and falls back to helm. Also documents the no-limit
pagination invariant and pins it with a test.
redis.py logged through stdlib logging with extra={"team": team}, which the
structlog bridge drops on the floor — the trap notify.py documents. Switched to
a bound logger with team as a kwarg.
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""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_of_deprovision_leaves_the_instance_deleting_to_be_retried(
|
|
pool: DictPool,
|
|
) -> None:
|
|
"""A dead-lettered deprovision must not strand the instance in `failed`.
|
|
|
|
The instance is `deleting`, which is one of the states that CAN legally become `failed`,
|
|
so the naive blanket UPDATE would move it there. due_for_deprovision only re-selects
|
|
`ready`(expired) and `deleting`, so `failed` would take the instance out of the recovery
|
|
sweep and leak the helm release forever. reconcile.py documents that a deprovision which
|
|
exhausts its retries stays re-enqueueable; this pins that guarantee.
|
|
"""
|
|
iid = await make_instance(pool, state=InstanceState.DELETING)
|
|
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, "cluster unreachable", "w1", max_attempts=5) is True
|
|
|
|
assert (await _task_row(pool, tid))["state"] == "failed"
|
|
inst = await instances.get(iid, team="platform")
|
|
assert inst is not None
|
|
assert inst.state is InstanceState.DELETING, "a stranded deprovision leaks the release"
|
|
assert inst.error is None
|
|
|
|
|
|
async def test_fail_of_upgrade_leaves_a_working_instance_ready(pool: DictPool) -> None:
|
|
"""A dead-lettered upgrade must not mark a healthy instance `failed`.
|
|
|
|
helm --atomic rolls the release back, so after a failed upgrade the instance is still
|
|
`ready` and serving the previous version. Marking it `failed` mislabels a working
|
|
service and drops it off the upgrade work-list. check_version_drift retries on the next
|
|
window; the dead-letter metric is the operator signal.
|
|
"""
|
|
iid = await make_instance(pool, state=InstanceState.READY)
|
|
tasks, instances = TaskRepo(pool), InstanceRepo(pool)
|
|
tid = await tasks.enqueue_standalone(iid, TaskKind.UPGRADE)
|
|
|
|
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, "upgrade to 1.4.0 kept timing out", "w1", max_attempts=5) is True
|
|
|
|
assert (await _task_row(pool, tid))["state"] == "failed"
|
|
inst = await instances.get(iid, team="platform")
|
|
assert inst is not None
|
|
assert inst.state is InstanceState.READY, "a failed upgrade mislabelled a healthy instance"
|
|
assert inst.error is None
|
|
|
|
|
|
async def test_fail_of_verify_leaves_the_instance_ready(pool: DictPool) -> None:
|
|
"""A dead-lettered verify must not mark the instance `failed`.
|
|
|
|
handle_verify halts the rollout for the service type; the instance itself is `ready`,
|
|
and drift re-provisions it if its release vanished. `failed` would take it out of both
|
|
recovery paths.
|
|
"""
|
|
iid = await make_instance(pool, state=InstanceState.READY)
|
|
tasks, instances = TaskRepo(pool), InstanceRepo(pool)
|
|
tid = await tasks.enqueue_standalone(iid, TaskKind.VERIFY)
|
|
|
|
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, "release vanished after upgrade", "w1", max_attempts=5) is True
|
|
|
|
assert (await _task_row(pool, tid))["state"] == "failed"
|
|
inst = await instances.get(iid, team="platform")
|
|
assert inst is not None
|
|
assert inst.state is InstanceState.READY
|
|
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"
|