fix: three correctness bugs found in review + a dropped-log

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.
This commit is contained in:
Nguyen Minh Phuc
2026-07-21 01:31:11 +00:00
parent d6c1b64512
commit d64c3c9f39
7 changed files with 299 additions and 31 deletions
@@ -278,12 +278,12 @@ class HelmProvisioner:
sweep. They are not orphans. They were never svcforge's to know about.
Cost is why this does not shell out to helm when it does not have to. `--selector` is
NOT pushed down as a server-side selector — helm fetches and decompresses every
release secret in the cluster regardless, then filters what it already parsed.
Measured: unscoped 23 releases in 4392ms, scoped to 0 in 3988ms, a saving of about
10%, not the order of magnitude the flag's shape suggests.
applied by helm after it has already fetched and decompressed every release secret in
the cluster, so it saves almost nothing: measured unscoped 23 releases in 4392ms,
scoped to 0 in 3988ms, about 10%. The flag's shape suggests a server-side filter; it
is a client-side one.
That was not academic. With the CPU request mutated to 0 by a cluster policy, the
The cost was real. With the CPU request mutated to 0 by a cluster policy, the
call took over 330s and timed out on every tick, against ~4s given real CPU. A check
that never completes reports no drift, which looks exactly like no drift existing.
@@ -347,19 +347,27 @@ class HelmProvisioner:
treating it as missing would have the reconciler re-provision on top of it.
* helm writes one secret per revision, so a release can still appear more than once
— 20 releases here had up to 10 revisions each. The newest `version` label wins.
Skipping this would report a release that exists as several, which for the
reconciler's set difference is harmless, and for anything counting releases is not.
Skipping this would report one release as several; the reconciler's set difference
tolerates that, but any caller that counts releases would over-count.
`chart`, `status`, `revision` and `app_version` on the returned ReleaseInfo are the
subset the labels give away free. `chart` in particular is empty rather than wrong,
because the chart name lives only in the compressed payload. Narrowing the model
instead would have been honest about this method and dishonest about the helm path,
which does populate them.
subset the labels give away free. `chart` is empty here because the chart name lives
only in the compressed payload. The model keeps those fields so the helm fallback
path, which does populate them from `helm list`, returns the same shape.
"""
try:
token = _SA_TOKEN.read_text(encoding="utf-8").strip()
except OSError:
return None
# The CA has to be readable too, and it is checked here rather than left to httpx.
# httpx loads the CA eagerly when the client is built, and that load raises OSError,
# which is not in the (httpx.HTTPError, json.JSONDecodeError) except below. A
# half-mounted ServiceAccount — token present, ca.crt absent or late — would then
# crash the tick as a bare bug instead of falling back. A complete ServiceAccount is
# the real in-cluster signal, so a missing CA means "not in-cluster" like a missing
# token does.
if not token or not os.access(_SA_CA, os.R_OK):
return None
host, port = (
os.environ.get("KUBERNETES_SERVICE_HOST"),
os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443"),
@@ -370,6 +378,12 @@ class HelmProvisioner:
selector = f"{_HELM_OWNER_LABEL},{MANAGED_BY_LABEL}={MANAGED_BY_VALUE},{_LIVE_STATUSES}"
try:
async with httpx.AsyncClient(verify=str(_SA_CA), timeout=_API_TIMEOUT_S) as client:
# No `limit` param, and that is load-bearing: the apiserver only returns a
# `metadata.continue` token when the client sets `limit`, so with none set it
# returns the full matching set in one response and the single read below is
# complete. Adding `limit` here without also looping on `continue` would
# silently truncate the list, and the reconciler would read the missing
# releases as orphans to delete or as vanished releases to re-provision.
resp = await client.get(
f"https://{host}:{port}/api/v1/secrets",
params={"labelSelector": selector},
@@ -39,7 +39,6 @@ you cared about.
from __future__ import annotations
import asyncio
import logging
import math
from dataclasses import dataclass
from datetime import UTC, datetime
@@ -51,6 +50,7 @@ from pydantic import ValidationError
from redis.asyncio import Redis
from redis.exceptions import RedisError
from svcforge_core import obs
from svcforge_core.adapters.clock import Clock, SystemClock
from svcforge_core.domain.models import Instance
@@ -59,7 +59,10 @@ if TYPE_CHECKING:
from svcforge_core.settings import Settings
_log = logging.getLogger(__name__)
# structlog via obs, not stdlib logging. The stdlib bridge builds the event dict from the
# record message alone and drops `extra=` fields — the same trap notify.py documents. A bound
# logger takes fields as kwargs (team=team) and keeps them. Bound per instance in __init__,
# which runs after obs.setup() has configured structlog, never at import time.
# --- The budget metric ------------------------------------------------------------------
#
@@ -209,6 +212,7 @@ class RateLimiter:
def __init__(self, r: Redis, limit: int, window_s: int, *, clock: Clock | None = None) -> None:
"""`clock` is injectable so the window boundary is testable without sleeping."""
self._log = obs.get_logger(__name__)
if limit < 1:
raise ValueError("limit must be >= 1")
if window_s < 1:
@@ -244,10 +248,10 @@ class RateLimiter:
allowed, remaining = await self._script(keys=[key], args=[self._limit, self._window_s])
except _REDIS_DOWN:
REDIS_ERRORS.labels(op="ratelimit").inc()
_log.warning(
self._log.warning(
"rate limiter degraded: redis unavailable, failing OPEN",
exc_info=True,
extra={"team": team},
team=team,
)
return RateLimitResult(
allowed=True,
@@ -294,6 +298,7 @@ class IdempotencyStore:
def __init__(self, r: Redis, ttl_s: int = 86400) -> None:
"""A day is the window a client might reasonably retry in; then the key is garbage."""
self._log = obs.get_logger(__name__)
if ttl_s < 1:
raise ValueError("ttl_s must be >= 1")
self._r = r
@@ -320,7 +325,7 @@ class IdempotencyStore:
existing = await self._r.get(redis_key)
except _REDIS_DOWN:
REDIS_ERRORS.labels(op="idempotency").inc()
_log.warning(
self._log.warning(
"idempotency degraded: redis unavailable, falling through to the DB constraint",
exc_info=True,
)
@@ -333,7 +338,7 @@ class IdempotencyStore:
try:
return UUID(_as_text(existing))
except ValueError:
_log.warning("idempotency key holds a non-UUID value; ignoring it")
self._log.warning("idempotency key holds a non-UUID value; ignoring it")
return None
@@ -371,6 +376,7 @@ class InstanceCache:
"""
def __init__(self, r: Redis, ttl_s: int = 30) -> None:
self._log = obs.get_logger(__name__)
if ttl_s < 1:
raise ValueError("ttl_s must be >= 1")
self._r = r
@@ -391,7 +397,7 @@ class InstanceCache:
raw = await self._r.get(self._key(instance_id))
except _REDIS_DOWN:
REDIS_ERRORS.labels(op="cache_get").inc()
_log.warning("cache read degraded: redis unavailable, falling through to Postgres")
self._log.warning("cache read degraded: redis unavailable, falling through to Postgres")
return None
if raw is None:
return None
@@ -400,7 +406,7 @@ class InstanceCache:
except ValidationError:
# A model change deployed over a warm cache. Treat it as a miss and let the TTL
# take the old shape out. Not an error: the truth is in Postgres either way.
_log.info("cache entry failed validation; treating as a miss")
self._log.info("cache entry failed validation; treating as a miss")
return None
async def put(self, inst: Instance) -> None:
@@ -410,7 +416,7 @@ class InstanceCache:
await self._r.set(self._key(inst.id), inst.model_dump_json(), ex=self._ttl_s)
except _REDIS_DOWN:
REDIS_ERRORS.labels(op="cache_put").inc()
_log.warning("cache write degraded: redis unavailable")
self._log.warning("cache write degraded: redis unavailable")
async def invalidate(self, instance_id: UUID) -> None:
"""One DEL. Called by the worker inside the code path that writes the state.
@@ -423,4 +429,4 @@ class InstanceCache:
await self._r.delete(self._key(instance_id))
except _REDIS_DOWN:
REDIS_ERRORS.labels(op="cache_del").inc()
_log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL")
self._log.warning("cache invalidate degraded: redis unavailable; entry expires within the TTL")
+22 -8
View File
@@ -215,14 +215,28 @@ class TaskRepo:
where id = %s""",
(err[-2000:], task_id),
)
# `state = any(%s)` keeps this honest: a deprovision that exhausts its
# retries against an already-deleted instance records nothing rather than
# resurrecting it into `failed`.
await cur.execute(
"""update instances set error=%s, state=%s, updated_at=now()
where id=%s and state = any(%s)""",
(err[-2000:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)),
)
# Dead-lettering the task is correct for every kind. Moving the INSTANCE to
# `failed` is correct only for provision: a provisioning instance that never
# came up is failed, and nothing recovers it but a human. The other kinds
# must leave the instance where it is, because for each of them the instance
# is still healthy and something else is responsible for recovery:
# deprovision — still `deleting`, which is exactly what lets
# due_for_deprovision re-enqueue it on the next sweep. `failed`
# drops it out of that query and leaks the release forever.
# upgrade — helm --atomic rolled back, so it is still `ready` and
# serving the previous version. check_version_drift retries on
# the next window; `failed` would mislabel a working service
# and drop it off the upgrade work-list.
# verify — handle_verify already halted the rollout; the instance is
# `ready`, and drift re-provisions it if its release vanished.
# The dead-letter metric and its alert are the operator signal for all four,
# so leaving the instance alone loses no visibility.
if row["kind"] == TaskKind.PROVISION.value:
await cur.execute(
"""update instances set error=%s, state=%s, updated_at=now()
where id=%s and state = any(%s)""",
(err[-2000:], InstanceState.FAILED.value, instance_id, list(_CAN_FAIL)),
)
# Counted here, not in the worker: this is the only place that knows the
# difference between "attempt 2 of 5 failed" and "this task is done trying".
TASKS_DEAD_LETTERED.labels(kind=str(row["kind"])).inc()
+26 -1
View File
@@ -11,9 +11,11 @@ import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from jwt import PyJWKClient
from starlette.exceptions import HTTPException
from services.api.models import ErrorBody
from services.api.routes import health, instances
@@ -89,6 +91,13 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo
Handlers raise `detail={"code": ..., "message": ...}`; FastAPI's default would nest
that under `{"detail": {...}}`. Plain-string details (raised by FastAPI itself, e.g.
a 405) are wrapped so clients never have to branch on the body's type.
Registered on starlette's HTTPException, not fastapi's. fastapi.HTTPException is a
subclass, and Starlette matches handlers by walking type(exc).__mro__, so a handler
keyed on the subclass never fires for a framework-raised 404 or 405 — which are
starlette.HTTPException instances. Keying on the parent catches both: app handlers
raise the FastAPI subclass with a dict detail, the framework raises the parent with a
str detail, and the branch below renders each into ErrorBody.
"""
assert isinstance(exc, HTTPException) # noqa: S101 - registered only for HTTPException
# Widened to object deliberately. Starlette types `detail` as str, but FastAPI passes
@@ -102,6 +111,21 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo
return JSONResponse(status_code=exc.status_code, content=body.model_dump(), headers=exc.headers)
async def _validation_exception_handler(request: Request, exc: Exception) -> JSONResponse:
"""Render request-validation failures as ErrorBody too.
A body that fails validation (a forbidden extra field, a bad type, an out-of-range
ttl_days) raises RequestValidationError, which the HTTPException handler above never
sees. Without this it returns FastAPI's default `{"detail": [...]}` — a second 422 shape
alongside the ErrorBody 422s the handlers raise. This gives every 422 one shape.
"""
assert isinstance(exc, RequestValidationError) # noqa: S101 - registered only for this
return JSONResponse(
status_code=422,
content=ErrorBody(code="validation_error", message=str(exc.errors())).model_dump(),
)
def create_app(settings: Settings | None = None) -> FastAPI:
"""App factory: lifespan, routers, exception handler, /metrics."""
settings = settings or load_settings()
@@ -132,6 +156,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
app.include_router(instances.router)
app.add_exception_handler(HTTPException, _http_exception_handler)
app.add_exception_handler(RequestValidationError, _validation_exception_handler)
return app
+40
View File
@@ -298,6 +298,46 @@ async def test_ttl_out_of_range_is_422(client: httpx.AsyncClient, token: str) ->
assert resp.status_code == 422
def _is_error_body(payload: object) -> bool:
"""The uniform error shape: a dict with `code` and `message`, and no default `detail`."""
return isinstance(payload, dict) and "code" in payload and "message" in payload
async def test_framework_404_uses_the_error_body_shape(client: httpx.AsyncClient) -> None:
"""A 404 raised by the router, not a handler, must still be ErrorBody.
Starlette raises its own HTTPException for an unknown route. The exception handler is
registered on that parent class precisely so this body is ErrorBody and not FastAPI's
default `{"detail": "Not Found"}` — one shape for every error.
"""
resp = await client.get("/v1/no-such-route")
assert resp.status_code == 404
assert _is_error_body(resp.json()), resp.text
async def test_framework_405_uses_the_error_body_shape(client: httpx.AsyncClient) -> None:
"""A wrong-method 405 comes from the router too, and must be ErrorBody."""
resp = await client.delete("/v1/instances") # collection route has no DELETE
assert resp.status_code == 405
assert _is_error_body(resp.json()), resp.text
async def test_body_validation_422_uses_the_error_body_shape(client: httpx.AsyncClient, token: str) -> None:
"""A RequestValidationError 422 must match the handler-raised 422 shape.
A forbidden extra field trips pydantic's `extra="forbid"` and raises
RequestValidationError, which the dedicated handler renders as ErrorBody rather than the
default `{"detail": [...]}`.
"""
resp = await client.post(
"/v1/instances",
headers=auth(token),
json={"service_type": "redis", "size": "small", "surprise": "field"},
)
assert resp.status_code == 422
assert _is_error_body(resp.json()), resp.text
# --------------------------------------------------------------------------- authn
+80
View File
@@ -110,6 +110,86 @@ async def test_fail_does_not_resurrect_a_deleted_instance(pool: DictPool) -> Non
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)
+90 -1
View File
@@ -21,6 +21,7 @@ from svcforge_core.adapters import helm
from svcforge_core.adapters.helm import (
MANAGED_BY_LABEL,
MANAGED_BY_VALUE,
HelmError,
HelmProvisioner,
)
from svcforge_core.domain.models import CatalogEntry, SizeSpec
@@ -131,8 +132,10 @@ def _fake_api(
token = tmp_path / "token"
token.write_text("tok", encoding="utf-8")
ca = tmp_path / "ca.crt"
ca.write_text("ca", encoding="utf-8") # a complete SA has both; the readability check needs it
monkeypatch.setattr(helm, "_SA_TOKEN", token)
monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
monkeypatch.setattr(helm, "_SA_CA", ca)
monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
monkeypatch.setenv("KUBERNETES_SERVICE_PORT_HTTPS", "443")
@@ -267,3 +270,89 @@ async def test_api_read_handles_kubernetes_serialising_empty_as_null(
monkeypatch.setattr(httpx, "AsyncClient", _NullClient)
assert await HelmProvisioner().list_releases() == []
@pytest.mark.asyncio
async def test_api_read_passes_tls_verify_and_timeout(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""The API client must verify against the SA CA and carry the read timeout.
A refactor that dropped `verify` to the default (or None) is a TLS regression the happy
path would not reveal, so it is pinned here off the captured client kwargs.
"""
cap = _fake_api(monkeypatch, tmp_path, [])
await HelmProvisioner().list_releases()
assert cap["client_kwargs"]["verify"] == str(helm._SA_CA)
assert cap["client_kwargs"]["timeout"] == helm._API_TIMEOUT_S
@pytest.mark.asyncio
async def test_api_read_sends_no_limit_param(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""No `limit`, so the apiserver returns the full set and the single read is complete.
Pins the pagination invariant: adding `limit` without consuming `metadata.continue`
would silently truncate the release list.
"""
cap = _fake_api(monkeypatch, tmp_path, [])
await HelmProvisioner().list_releases()
assert "limit" not in cap["params"]
@pytest.mark.asyncio
async def test_api_error_raises_helmerror_and_does_not_fall_back(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A reachable-but-erroring apiserver raises HelmError; it must not shell out to helm.
Falling back would swap a visible error for the 330s helm timeout this method exists to
remove. The fallback is only for a ServiceAccount that is not present at all.
"""
(tmp_path / "ca.crt").write_text("ca", encoding="utf-8")
monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token")
(tmp_path / "token").write_text("tok", encoding="utf-8")
monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
seen = _capture(monkeypatch)
class _ErrClient:
def __init__(self, **kw: object) -> None:
pass
async def __aenter__(self) -> _ErrClient:
return self
async def __aexit__(self, *exc: object) -> None:
return None
async def get(self, url: str, **kw: object) -> object:
raise httpx.ConnectError("connection reset by peer")
monkeypatch.setattr(httpx, "AsyncClient", _ErrClient)
with pytest.raises(HelmError):
await HelmProvisioner().list_releases()
assert seen == [], "an API error must not fall back to `helm list`"
@pytest.mark.asyncio
async def test_missing_ca_falls_back_to_helm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""A token without a readable CA is a half-mounted SA: fall back rather than crash.
httpx loads the CA when the client is built, raising OSError that the API except clause
does not catch, so the readability check has to happen before the request. A missing CA
means "not in-cluster", the same as a missing token.
"""
monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "token")
(tmp_path / "token").write_text("tok", encoding="utf-8")
monkeypatch.setattr(helm, "_SA_CA", tmp_path / "absent-ca.crt") # never created
monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
seen = _capture(monkeypatch)
await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases()
assert seen and seen[0][:2] == ["helm", "list"]