From 66eb6cb0eeda58557bf5985db8784e3cc42cb8c8 Mon Sep 17 00:00:00 2001 From: Nguyen Minh Phuc Date: Tue, 21 Jul 2026 01:42:25 +0000 Subject: [PATCH] refactor: converge the patterns multiple authors left divergent The codebase was written by several agents and had the same concept done more than one way. This makes it read as one voice, with no behaviour change. Dedup, each to a single canonical form: - INSTANCE_COLUMNS: the 13-column instances SELECT list existed as _COLUMNS in instances.py and reconcile.py (byte-identical) and inlined a third time in the worker. One exported constant now. - Settings.runtime_dsn: the three entrypoints each chose between str(pg_dsn) and pg_dsn.unicode_string(). One property. - yaml_tempfile: helm._ValuesFile and k8s._ManifestFile were the same write-yaml-to-a-temp-dir context manager. One helper in adapters/tempyaml.py. - services/_runtime.py: sleep_or_stop and install_stop_signals were copied between the worker and reconciler loops. One module, so shutdown behaviour cannot drift between them. - k8s.ensure_namespace used MANAGED_BY_LABEL/VALUE from helm.py instead of a hardcoded literal, so the managed-by label has one definition. - SvcforgeError is now the root of every svcforge exception (CatalogError, IllegalTransition, BadWindow, HandlerError), keeping each stdlib base in the MRO, so `except SvcforgeError` means what errors.py says it does. - ERROR_MAX_CHARS replaces the repeated `[-2000:]` truncation feeding the same error columns. - the reconciler reads settings.metrics_port like the worker, dropping its duplicate DEFAULT_METRICS_PORT and redundant --metrics-port option; the SVCFORGE_METRICS_PORT env override still applies through pydantic. Two smaller correctness/consistency fixes: - RateLimitResult.retry_after_s computed its delta against datetime.now(UTC) while the limiter runs on an injectable clock, so it was meaningless under a FakeClock and drifted by request latency in production. It now carries a checked_at from the same clock as reset_at. - handle_provision's notifier.send is wrapped like the reconciler's: a flaky webhook after the READY CAS would fail the task, and the retry would hit the READY early-return and drop the notification, turning a good provision into a failed one. --- .../svcforge_core/adapters/helm.py | 32 +-------------- .../svcforge_core/adapters/k8s.py | 33 ++------------- .../svcforge_core/adapters/redis.py | 8 +++- .../svcforge_core/adapters/tempyaml.py | 34 ++++++++++++++++ .../svcforge_core/domain/catalog.py | 3 +- .../svcforge_core/domain/states.py | 4 +- .../svcforge_core/domain/windows.py | 4 +- libs/svcforge_core/svcforge_core/repo/db.py | 5 +++ .../svcforge_core/repo/instances.py | 12 +++--- .../svcforge_core/repo/reconcile.py | 12 +++--- .../svcforge_core/svcforge_core/repo/tasks.py | 8 ++-- libs/svcforge_core/svcforge_core/settings.py | 9 +++++ services/_runtime.py | 37 +++++++++++++++++ services/api/main.py | 2 +- services/reconciler/main.py | 40 +++++-------------- services/worker/handlers.py | 29 +++++++++----- services/worker/main.py | 27 +++---------- tests/fakes.py | 8 +++- tests/integration/test_redis.py | 2 +- 19 files changed, 166 insertions(+), 143 deletions(-) create mode 100644 libs/svcforge_core/svcforge_core/adapters/tempyaml.py create mode 100644 services/_runtime.py diff --git a/libs/svcforge_core/svcforge_core/adapters/helm.py b/libs/svcforge_core/svcforge_core/adapters/helm.py index abef23b..9f1f76d 100644 --- a/libs/svcforge_core/svcforge_core/adapters/helm.py +++ b/libs/svcforge_core/svcforge_core/adapters/helm.py @@ -21,17 +21,15 @@ from __future__ import annotations import asyncio import json import os -import shutil import signal -import tempfile from collections.abc import Sequence from pathlib import Path from typing import Any, Protocol import httpx -import yaml from pydantic import BaseModel, ConfigDict, Field +from svcforge_core.adapters.tempyaml import yaml_tempfile from svcforge_core.domain.models import CatalogEntry from svcforge_core.errors import SvcforgeError @@ -220,7 +218,7 @@ class HelmProvisioner: # `--wait` is why `ready` in the DB means ready — it returns when the pods are up. # `--atomic` rolls back a failed upgrade; it doubles the worst case, which is what # `_RUN_TIMEOUT_MARGIN_S` and helm's own `--timeout` are sized around. - with _values_file(values) as path: + with yaml_tempfile(values, prefix="svcforge-values-", name="values.yaml") as path: argv = self._base_argv( "upgrade", "--install", @@ -433,29 +431,3 @@ class HelmProvisioner: ), ) return [info for _, info in newest.values()] - - -class _ValuesFile: - """Context manager yielding a path to a values.yaml written from a dict. - - A file, not `--set`: `--set` has its own escaping grammar (commas, dots, backslashes) and - values carry tenant-shaped strings. Serialising YAML sidesteps the grammar entirely. - """ - - def __init__(self, values: dict[str, Any]) -> None: - self._values = values - self._dir: str | None = None - - def __enter__(self) -> Path: - self._dir = tempfile.mkdtemp(prefix="svcforge-values-") - path = Path(self._dir) / "values.yaml" - path.write_text(yaml.safe_dump(self._values, default_flow_style=False), encoding="utf-8") - return path - - def __exit__(self, *exc: object) -> None: - if self._dir is not None: - shutil.rmtree(self._dir, ignore_errors=True) - - -def _values_file(values: dict[str, Any]) -> _ValuesFile: - return _ValuesFile(values) diff --git a/libs/svcforge_core/svcforge_core/adapters/k8s.py b/libs/svcforge_core/svcforge_core/adapters/k8s.py index fdcd752..75ae139 100644 --- a/libs/svcforge_core/svcforge_core/adapters/k8s.py +++ b/libs/svcforge_core/svcforge_core/adapters/k8s.py @@ -13,14 +13,11 @@ from __future__ import annotations import base64 import binascii import json -import shutil -import tempfile from pathlib import Path from typing import Any -import yaml - -from svcforge_core.adapters.helm import HelmError, _run +from svcforge_core.adapters.helm import MANAGED_BY_LABEL, MANAGED_BY_VALUE, HelmError, _run +from svcforge_core.adapters.tempyaml import yaml_tempfile from svcforge_core.errors import SvcforgeError _KUBECTL_TIMEOUT_S = 60 @@ -71,10 +68,10 @@ class KubectlClient: "kind": "Namespace", "metadata": { "name": ns, - "labels": {"app.kubernetes.io/managed-by": "svcforge", **(labels or {})}, + "labels": {MANAGED_BY_LABEL: MANAGED_BY_VALUE, **(labels or {})}, }, } - with _manifest_file(manifest) as path: + with yaml_tempfile(manifest, prefix="svcforge-manifest-", name="manifest.yaml") as path: await self._kubectl("apply", "--filename", str(path)) async def read_secret(self, ns: str, name: str) -> dict[str, str]: @@ -122,25 +119,3 @@ class KubectlClient: # timeout path does not come through HelmError. Without this clause a wedged # kubectl surfaces as TimeoutError past a caller written to `except K8sError`. raise K8sError(f"kubectl {args[0] if args else ''} timed out after {self._timeout_s}s") from exc - - -class _ManifestFile: - """A temp file holding one YAML manifest, removed on exit.""" - - def __init__(self, manifest: dict[str, Any]) -> None: - self._manifest = manifest - self._dir: str | None = None - - def __enter__(self) -> Path: - self._dir = tempfile.mkdtemp(prefix="svcforge-manifest-") - path = Path(self._dir) / "manifest.yaml" - path.write_text(yaml.safe_dump(self._manifest, default_flow_style=False), encoding="utf-8") - return path - - def __exit__(self, *exc: object) -> None: - if self._dir is not None: - shutil.rmtree(self._dir, ignore_errors=True) - - -def _manifest_file(manifest: dict[str, Any]) -> _ManifestFile: - return _ManifestFile(manifest) diff --git a/libs/svcforge_core/svcforge_core/adapters/redis.py b/libs/svcforge_core/svcforge_core/adapters/redis.py index d2a3d90..ebf7035 100644 --- a/libs/svcforge_core/svcforge_core/adapters/redis.py +++ b/libs/svcforge_core/svcforge_core/adapters/redis.py @@ -178,6 +178,10 @@ class RateLimitResult: limit: int remaining: int reset_at: datetime + # When the limiter made this decision, from the same injected clock as reset_at. The two + # have to share a clock or retry_after_s (their difference) is meaningless under a + # FakeClock, and drifts by the request latency even in production. + checked_at: datetime degraded: bool = False @property @@ -187,7 +191,7 @@ class RateLimitResult: Rounded up and floored at one: `Retry-After: 0` invites an immediate retry into the same closed window, which is a busy loop with extra steps. """ - delta = (self.reset_at - datetime.now(UTC)).total_seconds() + delta = (self.reset_at - self.checked_at).total_seconds() return max(1, math.ceil(delta)) @@ -258,6 +262,7 @@ class RateLimiter: limit=self._limit, remaining=self._limit, reset_at=reset_at, + checked_at=self._clock.now(), degraded=True, ) return RateLimitResult( @@ -265,6 +270,7 @@ class RateLimiter: limit=self._limit, remaining=int(remaining), reset_at=reset_at, + checked_at=self._clock.now(), ) diff --git a/libs/svcforge_core/svcforge_core/adapters/tempyaml.py b/libs/svcforge_core/svcforge_core/adapters/tempyaml.py new file mode 100644 index 0000000..6e38bf8 --- /dev/null +++ b/libs/svcforge_core/svcforge_core/adapters/tempyaml.py @@ -0,0 +1,34 @@ +"""One temp YAML file, written from a dict and removed on exit. + +helm and kubectl both take their input as a file rather than on the command line: `--set` +and inline manifests each have their own escaping grammar, and tenant-shaped values would +have to be escaped into it. Serialising YAML to a file sidesteps the grammar entirely. Both +adapters needed the same throwaway-file dance, so it lives here once. +""" + +from __future__ import annotations + +import shutil +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Any + +import yaml + + +@contextmanager +def yaml_tempfile(payload: dict[str, Any], *, prefix: str, name: str) -> Iterator[Path]: + """Yield a path to `name` inside a fresh temp dir, holding `payload` as YAML. + + The whole dir is removed on exit, `ignore_errors` so a cleanup race never masks the real + error from the block. + """ + tmpdir = tempfile.mkdtemp(prefix=prefix) + try: + path = Path(tmpdir) / name + path.write_text(yaml.safe_dump(payload, default_flow_style=False), encoding="utf-8") + yield path + finally: + shutil.rmtree(tmpdir, ignore_errors=True) diff --git a/libs/svcforge_core/svcforge_core/domain/catalog.py b/libs/svcforge_core/svcforge_core/domain/catalog.py index cd66cb7..95c3178 100644 --- a/libs/svcforge_core/svcforge_core/domain/catalog.py +++ b/libs/svcforge_core/svcforge_core/domain/catalog.py @@ -10,9 +10,10 @@ import yaml from pydantic import ValidationError from svcforge_core.domain.models import CatalogEntry +from svcforge_core.errors import SvcforgeError -class CatalogError(Exception): +class CatalogError(SvcforgeError): """A catalog file could not be parsed or validated. `key` names the offending service type, or None when the failure is file-level. diff --git a/libs/svcforge_core/svcforge_core/domain/states.py b/libs/svcforge_core/svcforge_core/domain/states.py index 61dea7b..d0c1120 100644 --- a/libs/svcforge_core/svcforge_core/domain/states.py +++ b/libs/svcforge_core/svcforge_core/domain/states.py @@ -3,6 +3,8 @@ from enum import StrEnum from typing import Final +from svcforge_core.errors import SvcforgeError + class InstanceState(StrEnum): """Lifecycle of a provisioned service instance.""" @@ -15,7 +17,7 @@ class InstanceState(StrEnum): FAILED = "failed" -class IllegalTransition(Exception): +class IllegalTransition(SvcforgeError): """Raised by transition() when cur -> nxt is not in LEGAL.""" diff --git a/libs/svcforge_core/svcforge_core/domain/windows.py b/libs/svcforge_core/svcforge_core/domain/windows.py index 46ea52f..21dac8a 100644 --- a/libs/svcforge_core/svcforge_core/domain/windows.py +++ b/libs/svcforge_core/svcforge_core/domain/windows.py @@ -19,11 +19,13 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from croniter import croniter +from svcforge_core.errors import SvcforgeError + CRON_FIELDS = 5 SEPARATOR = "|" -class BadWindow(ValueError): +class BadWindow(SvcforgeError, ValueError): """A maintenance window spec is not a 5-field cron plus a known IANA zone.""" diff --git a/libs/svcforge_core/svcforge_core/repo/db.py b/libs/svcforge_core/svcforge_core/repo/db.py index 63df47e..8215337 100644 --- a/libs/svcforge_core/svcforge_core/repo/db.py +++ b/libs/svcforge_core/svcforge_core/repo/db.py @@ -17,6 +17,11 @@ from psycopg_pool import AsyncConnectionPool # against a bare `AsyncConnectionPool`, which resolves to tuple rows. The runtime was # always right; without these aliases the annotations quietly disagree with it, and the # fix people reach for is `# type: ignore`, which throws away the checking entirely. +# The cap on error text written to the `instances.error` and `tasks.last_error` columns. A +# helm failure can emit megabytes; these columns are read by humans. Defined once here rather +# than as a bare 2000 at each write, so the two call paths that feed the same columns agree. +ERROR_MAX_CHARS = 2000 + type DictRow = dict[str, Any] type DictConnection = AsyncConnection[DictRow] type DictPool = AsyncConnectionPool[DictConnection] diff --git a/libs/svcforge_core/svcforge_core/repo/instances.py b/libs/svcforge_core/svcforge_core/repo/instances.py index 415aa22..edb55d1 100644 --- a/libs/svcforge_core/svcforge_core/repo/instances.py +++ b/libs/svcforge_core/svcforge_core/repo/instances.py @@ -20,7 +20,7 @@ from svcforge_core.domain.models import Instance from svcforge_core.domain.states import InstanceState from svcforge_core.repo.db import DictPool -_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version, +INSTANCE_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version, endpoint, error, expires_at, created_at, updated_at""" @@ -57,7 +57,7 @@ class InstanceRepo: release_name, chart_version, endpoint, error, expires_at) values (%(id)s, %(team)s, %(service_type)s, %(size)s, %(state)s, %(namespace)s, %(release_name)s, %(chart_version)s, %(endpoint)s, %(error)s, %(expires_at)s) - returning {_COLUMNS}""", # noqa: S608 - _COLUMNS is a module constant, not input + returning {INSTANCE_COLUMNS}""", # noqa: S608 - INSTANCE_COLUMNS is a module constant, not input { "id": inst.id, "team": inst.team, @@ -80,7 +80,7 @@ class InstanceRepo: """Fetch one instance owned by `team`. None if it does not exist OR is not theirs.""" async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( - f"select {_COLUMNS} from instances where id = %s and team = %s", # noqa: S608 + f"select {INSTANCE_COLUMNS} from instances where id = %s and team = %s", # noqa: S608 (id, team), ) row = await cur.fetchone() @@ -90,7 +90,7 @@ class InstanceRepo: """The team's instances, newest first.""" async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( - f"""select {_COLUMNS} from instances + f"""select {INSTANCE_COLUMNS} from instances where team = %s order by created_at desc limit %s""", # noqa: S608 (team, limit), ) @@ -157,7 +157,7 @@ class InstanceRepo: # and would otherwise type these rows as tuples. async with self._pool.connection() as conn, conn.cursor(row_factory=dict_row) as cur: await cur.execute( - f"""select {_COLUMNS}, maintenance_window from instances + f"""select {INSTANCE_COLUMNS}, maintenance_window from instances where state = 'ready' and service_type = %(service_type)s and chart_version <> %(catalog_version)s @@ -166,7 +166,7 @@ class InstanceRepo: where cv.service_type = %(service_type)s and cv.rollout_state = 'halted') order by team = %(own_team)s desc, created_at - limit %(max_in_flight)s""", # noqa: S608 - _COLUMNS is a module constant, not input + limit %(max_in_flight)s""", # noqa: S608 - INSTANCE_COLUMNS is a module constant, not input { "service_type": service_type, "catalog_version": catalog_version, diff --git a/libs/svcforge_core/svcforge_core/repo/reconcile.py b/libs/svcforge_core/svcforge_core/repo/reconcile.py index 0665aad..04ed9da 100644 --- a/libs/svcforge_core/svcforge_core/repo/reconcile.py +++ b/libs/svcforge_core/svcforge_core/repo/reconcile.py @@ -25,10 +25,8 @@ from psycopg import AsyncCursor from svcforge_core.domain.models import Instance, TaskKind, TaskState from svcforge_core.domain.states import InstanceState, transition from svcforge_core.obs import inject_traceparent -from svcforge_core.repo.db import DictPool - -_COLUMNS = """id, team, service_type, size, state, namespace, release_name, chart_version, - endpoint, error, expires_at, created_at, updated_at""" +from svcforge_core.repo.db import ERROR_MAX_CHARS, DictPool +from svcforge_core.repo.instances import INSTANCE_COLUMNS # A task nobody will ever run again. The idempotency guard on every enqueue below asks # "is one already outstanding?", and 'done'/'failed' are not outstanding: a failed @@ -70,7 +68,7 @@ class ReconcileRepo: """Every instance the DB believes is running. The drift check's expectation.""" async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( - f"select {_COLUMNS} from instances where state = %s", # noqa: S608 - module constant + f"select {INSTANCE_COLUMNS} from instances where state = %s", # noqa: S608 - module constant (InstanceState.READY.value,), ) rows = await cur.fetchall() @@ -126,7 +124,7 @@ class ReconcileRepo: await cur.execute( "update instances set state = %s, error = %s, updated_at = now() where id = %s", - (provisioning.value, reason[-2000:], instance_id), + (provisioning.value, reason[-ERROR_MAX_CHARS:], instance_id), ) return await _insert_task(cur, instance_id, TaskKind.PROVISION) @@ -151,7 +149,7 @@ class ReconcileRepo: """ async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( - f"""select {_COLUMNS} from instances i + f"""select {INSTANCE_COLUMNS} from instances i where ((i.state = %(ready)s and i.expires_at < now()) or i.state = %(deleting)s) and not exists ( select 1 from tasks t diff --git a/libs/svcforge_core/svcforge_core/repo/tasks.py b/libs/svcforge_core/svcforge_core/repo/tasks.py index 4984034..64ee488 100644 --- a/libs/svcforge_core/svcforge_core/repo/tasks.py +++ b/libs/svcforge_core/svcforge_core/repo/tasks.py @@ -20,7 +20,7 @@ from svcforge_core.domain.backoff import next_attempt_at from svcforge_core.domain.models import Task, TaskKind from svcforge_core.domain.states import LEGAL, InstanceState from svcforge_core.obs import TASKS_DEAD_LETTERED, inject_traceparent -from svcforge_core.repo.db import DictPool +from svcforge_core.repo.db import ERROR_MAX_CHARS, DictPool # Which states may legally become `failed`, derived from the domain's own table rather # than restated here. Without this guard the UPDATE below would happily move a `deleted` @@ -205,7 +205,7 @@ class TaskRepo: set state='queued', locked_by=null, locked_at=null, last_error=%s, run_after=%s where id = %s""", - (err[-2000:], next_attempt_at(attempts - 1, now=now), task_id), + (err[-ERROR_MAX_CHARS:], next_attempt_at(attempts - 1, now=now), task_id), ) return True @@ -213,7 +213,7 @@ class TaskRepo: """update tasks set state='failed', locked_by=null, locked_at=null, last_error=%s where id = %s""", - (err[-2000:], task_id), + (err[-ERROR_MAX_CHARS:], task_id), ) # Dead-lettering the task is correct for every kind. Moving the INSTANCE to # `failed` is correct only for provision: a provisioning instance that never @@ -235,7 +235,7 @@ class TaskRepo: 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)), + (err[-ERROR_MAX_CHARS:], 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". diff --git a/libs/svcforge_core/svcforge_core/settings.py b/libs/svcforge_core/svcforge_core/settings.py index 6dfbac8..a55faa4 100644 --- a/libs/svcforge_core/svcforge_core/settings.py +++ b/libs/svcforge_core/svcforge_core/settings.py @@ -83,6 +83,15 @@ class Settings(BaseSettings): # just for /metrics. 9000 matches the chart's PodMonitor; change both or neither. metrics_port: int = Field(default=9000, ge=1, le=65535) + @property + def runtime_dsn(self) -> str: + """The transaction-pooler DSN the services open their pool against, as a string. + + A property so the three entrypoints do not each choose between `str(pg_dsn)` and + `pg_dsn.unicode_string()` — the two spellings that were drifting across the services. + """ + return str(self.pg_dsn) + @property def migration_dsn(self) -> str: """Migrations need a session-mode connection; fall back to the runtime DSN locally.""" diff --git a/services/_runtime.py b/services/_runtime.py new file mode 100644 index 0000000..dc703b0 --- /dev/null +++ b/services/_runtime.py @@ -0,0 +1,37 @@ +"""Shared asyncio scaffolding for the long-lived services. + +The worker and the reconciler are both a loop that runs until SIGTERM. They wake immediately +on shutdown rather than sleeping through it, and they install the same loop-safe signal +handlers. Both lived in each service before; keeping one copy means the shutdown behaviour +cannot drift between them. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import signal + + +async def sleep_or_stop(stop: asyncio.Event, seconds: float) -> None: + """Sleep for `seconds`, but return the instant `stop` is set. + + `await asyncio.sleep(seconds)` would make every SIGTERM cost up to `seconds` of + Kubernetes waiting on terminationGracePeriod for nothing. + """ + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(stop.wait(), timeout=seconds) + + +def install_stop_signals(stop: asyncio.Event) -> None: + """Set `stop` on SIGTERM and SIGINT, loop-safely. + + add_signal_handler, not signal.signal. signal.signal runs the handler at an arbitrary + bytecode boundary on whatever thread the C-level handler lands on, and the loop does not + notice until its next timer fires — up to a full sleep interval away. add_signal_handler + schedules the callback as an ordinary loop callback, so the sleep_or_stop above returns + at once. + """ + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stop.set) diff --git a/services/api/main.py b/services/api/main.py index a56767b..3a3340a 100644 --- a/services/api/main.py +++ b/services/api/main.py @@ -55,7 +55,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: RateLimiter(redis, limit=settings.rate_limit_per_minute, window_s=60) if redis is not None else None ) - pool = make_pool(str(settings.pg_dsn), settings.pool_min_size, settings.pool_max_size) + pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size) # wait=True fails NOW, loudly, if the DSN is wrong — instead of at the first request, # as a PoolTimeout, in front of a user. await pool.open(wait=True) diff --git a/services/reconciler/main.py b/services/reconciler/main.py index 4018b5c..b60b28f 100644 --- a/services/reconciler/main.py +++ b/services/reconciler/main.py @@ -23,19 +23,18 @@ Three rules hold the design together: binary that cannot reach the API server must not stop TTLs from expiring. * **Enqueue, never act.** The reconciler diagnoses; workers treat. It writes task rows and instance states, and never calls `helm install`. The one exception is reading — the drift - check runs `helm list`, because seeing reality is the job. + check lists the live releases, because seeing reality is the job. """ from __future__ import annotations import asyncio -import contextlib -import signal from collections.abc import Awaitable, Callable from dataclasses import dataclass import typer +from services._runtime import install_stop_signals, sleep_or_stop from svcforge_core.adapters.clock import Clock, SystemClock from svcforge_core.adapters.helm import HelmProvisioner, Provisioner from svcforge_core.adapters.notify import LogNotifier, Notifier @@ -59,9 +58,6 @@ from svcforge_core.settings import Settings, load_settings log = get_logger("svcforge.reconciler") -# The chart's PodMonitor scrapes the port named `metrics` on 9000. Keep them in step. -DEFAULT_METRICS_PORT = 9000 - @dataclass(frozen=True) class ReconcilerDeps: @@ -92,7 +88,7 @@ class ReconcilerDeps: async def check_drift(deps: ReconcilerDeps) -> None: - """`helm list -A -o json` versus what the database believes. + """The live helm releases versus what the database believes. This is the only check that looks outside Postgres, and the only one that can catch the failure nothing else can: someone ran `helm uninstall` by hand, or a node was drained @@ -312,12 +308,6 @@ async def _run_checks(deps: ReconcilerDeps) -> None: log.exception("gauges.failed") -async def _sleep_or_stop(stop: asyncio.Event, seconds: float) -> None: - """Sleep, but wake immediately on SIGTERM. A 60s nap must not cost 60s of shutdown.""" - with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=seconds) - - async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None: """Tick, sleep, repeat, until told to stop. @@ -328,7 +318,7 @@ async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None: """ while not stop.is_set(): await tick(deps) - await _sleep_or_stop(stop, deps.settings.reconcile_interval_s) + await sleep_or_stop(stop, deps.settings.reconcile_interval_s) def build_deps( @@ -353,11 +343,11 @@ def build_deps( ) -async def _amain(once: bool, metrics_port: int, own_team: str, max_in_flight: int) -> None: +async def _amain(once: bool, own_team: str, max_in_flight: int) -> None: settings = load_settings() setup("svcforge-reconciler", settings) - pool = make_pool(settings.pg_dsn.unicode_string(), settings.pool_min_size, settings.pool_max_size) + pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size) await pool.open(wait=True) deps = build_deps(pool, settings, own_team, max_in_flight) @@ -368,17 +358,12 @@ async def _amain(once: bool, metrics_port: int, own_team: str, max_in_flight: in await tick(deps) return - start_metrics_server(metrics_port) + # settings.metrics_port, like the worker. SVCFORGE_METRICS_PORT still overrides it, + # through pydantic rather than a second CLI option, so the port has one definition. + start_metrics_server(settings.metrics_port) stop = asyncio.Event() - loop = asyncio.get_running_loop() - for sig in (signal.SIGTERM, signal.SIGINT): - # add_signal_handler, NOT signal.signal. signal.signal fires the handler at an - # arbitrary bytecode boundary on the main thread and the loop does not notice - # until its next timer — which here is up to a full 60s tick away. This one is - # scheduled as an ordinary loop callback, so the `stop.wait()` above returns - # immediately. - loop.add_signal_handler(sig, stop.set) + install_stop_signals(stop) await run_reconciler(deps, stop) finally: @@ -391,9 +376,6 @@ app = typer.Typer(add_completion=False, help="svcforge reconciler: the control l @app.command() def main( once: bool = typer.Option(False, "--once", help="Run one tick and exit."), - metrics_port: int = typer.Option( - DEFAULT_METRICS_PORT, envvar="SVCFORGE_METRICS_PORT", help="Port for /metrics." - ), own_team: str = typer.Option( "platform", envvar="SVCFORGE_OWN_TEAM", help="Team whose instances upgrade first." ), @@ -403,7 +385,7 @@ def main( ) -> None: """Run the reconciler.""" # One asyncio.run, at the top, never nested. Everything below it is already async. - asyncio.run(_amain(once, metrics_port, own_team, max_in_flight)) + asyncio.run(_amain(once, own_team, max_in_flight)) if __name__ == "__main__": diff --git a/services/worker/handlers.py b/services/worker/handlers.py index fce124b..e8f78d3 100644 --- a/services/worker/handlers.py +++ b/services/worker/handlers.py @@ -18,18 +18,21 @@ from typing import Any from services.worker.deps import WorkerDeps from svcforge_core.domain.models import CatalogEntry, Instance, Task, TaskKind from svcforge_core.domain.states import InstanceState +from svcforge_core.errors import SvcforgeError +from svcforge_core.obs import get_logger +from svcforge_core.repo.instances import INSTANCE_COLUMNS + +log = get_logger("svcforge.worker") -class HandlerError(RuntimeError): +class HandlerError(SvcforgeError, RuntimeError): """A task failed in a way worth retrying. The message lands in tasks.last_error.""" async def _load_instance(task: Task, deps: WorkerDeps) -> Instance: async with deps.pool.connection() as conn, conn.cursor() as cur: await cur.execute( - """select id, team, service_type, size, state, namespace, release_name, - chart_version, endpoint, error, expires_at, created_at, updated_at - from instances where id = %s""", + f"select {INSTANCE_COLUMNS} from instances where id = %s", # noqa: S608 - module constant (task.instance_id,), ) row = await cur.fetchone() @@ -75,11 +78,19 @@ async def handle_provision(task: Task, deps: WorkerDeps) -> None: inst.id, InstanceState.PROVISIONING, InstanceState.READY, endpoint=endpoint ) if ok: - await deps.notifier.send( - "instance.ready", - f"instance {inst.id} is ready at {endpoint}", - {"instance_id": str(inst.id), "team": inst.team, "service_type": inst.service_type}, - ) + try: + await deps.notifier.send( + "instance.ready", + f"instance {inst.id} is ready at {endpoint}", + {"instance_id": str(inst.id), "team": inst.team, "service_type": inst.service_type}, + ) + except Exception: + # The provision succeeded and the row is already READY; the notification is a + # courtesy. Letting a webhook timeout propagate would fail the task, and the + # retry would hit the READY early-return and drop the notification anyway — so a + # flaky notifier would turn every provision into a "failed" task. Same guard the + # reconciler puts around its own notify. + log.exception("notify.failed", instance_id=str(inst.id)) async def handle_deprovision(task: Task, deps: WorkerDeps) -> None: diff --git a/services/worker/main.py b/services/worker/main.py index 6ea063e..f855ab4 100644 --- a/services/worker/main.py +++ b/services/worker/main.py @@ -10,13 +10,12 @@ for something better. from __future__ import annotations import asyncio -import contextlib -import signal import time from collections.abc import Awaitable from opentelemetry import trace +from services._runtime import install_stop_signals, sleep_or_stop from services.worker.deps import WorkerDeps from services.worker.handlers import HANDLERS from svcforge_core import obs @@ -33,16 +32,6 @@ from svcforge_core.settings import Settings, load_settings log = obs.get_logger("svcforge.worker") -async def _sleep_or_stop(stop: asyncio.Event, seconds: float) -> None: - """Sleep, but wake immediately on shutdown. - - `await asyncio.sleep(5)` would make every SIGTERM cost up to five seconds of - Kubernetes waiting on terminationGracePeriod for no reason. - """ - with contextlib.suppress(TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=seconds) - - async def _report(coro: Awaitable[bool], task_id: int, what: str) -> None: """Run a terminal report, and never let its failure escape. @@ -154,12 +143,12 @@ async def run_worker(deps: WorkerDeps, stop: asyncio.Event) -> None: # A DB blip must not kill the worker; back off and try again. log.exception("claim failed") sem.release() - await _sleep_or_stop(stop, deps.settings.poll_interval_s) + await sleep_or_stop(stop, deps.settings.poll_interval_s) continue if task is None: sem.release() - await _sleep_or_stop(stop, deps.settings.poll_interval_s) + await sleep_or_stop(stop, deps.settings.poll_interval_s) continue tg.create_task(_run_one(deps, task, sem)) @@ -176,7 +165,7 @@ async def _amain() -> None: settings.check_production() obs.start_metrics_server(settings.metrics_port) - pool = make_pool(settings.pg_dsn.unicode_string(), settings.pool_min_size, settings.pool_max_size) + pool = make_pool(settings.runtime_dsn, settings.pool_min_size, settings.pool_max_size) await pool.open(wait=True) deps = WorkerDeps( @@ -191,13 +180,7 @@ async def _amain() -> None: ) stop = asyncio.Event() - loop = asyncio.get_running_loop() - for sig in (signal.SIGTERM, signal.SIGINT): - # add_signal_handler, NOT signal.signal. signal.signal runs the handler at an - # arbitrary bytecode boundary on whatever thread the C-level handler lands on, - # and the event loop will not notice until its next timer fires. This one is - # loop-safe: the callback runs as a normal loop callback. - loop.add_signal_handler(sig, stop.set) + install_stop_signals(stop) try: await run_worker(deps, stop) diff --git a/tests/fakes.py b/tests/fakes.py index 48de2c7..73c340c 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -145,7 +145,12 @@ class FakeRateLimiter: # Fails OPEN, exactly like the real one. A limiter that refused here would make # "Redis is down" indistinguishable from "you are over quota". return RateLimitResult( - allowed=True, limit=self.limit, remaining=self.limit, reset_at=reset_at, degraded=True + allowed=True, + limit=self.limit, + remaining=self.limit, + reset_at=reset_at, + checked_at=self.clock.now(), + degraded=True, ) key = f"rl:{team}:{window}" n = self.counts.get(key, 0) + 1 @@ -155,6 +160,7 @@ class FakeRateLimiter: limit=self.limit, remaining=max(0, self.limit - n), reset_at=reset_at, + checked_at=self.clock.now(), ) diff --git a/tests/integration/test_redis.py b/tests/integration/test_redis.py index 4c28772..7cee970 100644 --- a/tests/integration/test_redis.py +++ b/tests/integration/test_redis.py @@ -434,7 +434,7 @@ async def test_the_budget_metric_is_exposed_and_labelled_by_op() -> None: """`curl -s localhost:8000/metrics | grep svcforge_redis_commands_total`.""" limiter = FakeRateLimiter(limit=10, window_s=60, clock=FakeClock(start=_T0)) await limiter.check("acme") # the fake does not touch the real counter - RateLimitResult(allowed=True, limit=10, remaining=9, reset_at=_T0) + RateLimitResult(allowed=True, limit=10, remaining=9, reset_at=_T0, checked_at=_T0) text = generate_latest(REGISTRY).decode()