From c53734d2bcf511bf58bd537c12b599db51202c20 Mon Sep 17 00:00:00 2001 From: Nguyen Minh Phuc Date: Tue, 21 Jul 2026 11:18:57 +0000 Subject: [PATCH] docs: add USER_GUIDE.md, tighten comments, fix CLI needing a DSN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The comment pass is prose-only: every distinct "why" is kept, the narration around it is not. Verified by AST-comparing each changed file against HEAD with docstrings stripped — only the two files below differ in executable code. Two real fixes fell out of the read-through: * The CLI documented itself as never touching the database, then called load_settings(), which requires SVCFORGE_PG_DSN. It refused to start without a Postgres URL it never opens. It now has its own two-field ClientSettings; the orphaned api_url/api_token are dropped from Settings, where nothing else read them. * repo/db.py had the DictRow alias comment and the ERROR_MAX_CHARS comment run together above the wrong symbol. USER_GUIDE.md is the caller-facing guide the README only gestured at: auth, catalog, every endpoint with curl, the lifecycle, the error table, rate limiting, the CLI, client generation, an end-to-end poll loop. It records two facts about the live deployment rather than documenting a flow nobody can run. SVCFORGE_JWKS_URL points at a realm with no IdP behind it, so the API logs "JWKS warm-up failed" at startup and every /v1 request is a 401. And `helm repo list` in the worker returns no repositories, so the three bitnamilegacy/ catalog entries cannot resolve at provision time; only the oci:// entries can. make lint clean, 76 unit + 111 integration tests pass. --- README.md | 25 +- USER_GUIDE.md | 267 ++++++++++++++++++ .../svcforge_core/adapters/clock.py | 19 +- .../svcforge_core/adapters/helm.py | 221 ++++++--------- .../svcforge_core/adapters/notify.py | 46 ++- .../svcforge_core/adapters/redis.py | 208 ++++++-------- libs/svcforge_core/svcforge_core/errors.py | 12 +- libs/svcforge_core/svcforge_core/obs.py | 122 ++++---- libs/svcforge_core/svcforge_core/repo/db.py | 44 ++- .../svcforge_core/repo/reconcile.py | 113 ++++---- .../svcforge_core/svcforge_core/repo/tasks.py | 144 ++++------ libs/svcforge_core/svcforge_core/runtime.py | 8 +- libs/svcforge_core/svcforge_core/settings.py | 24 +- services/api/deps.py | 49 ++-- services/api/main.py | 95 +++---- services/api/models.py | 14 +- services/api/routes/health.py | 38 ++- services/api/routes/instances.py | 58 ++-- services/cli/main.py | 28 +- services/reconciler/main.py | 175 ++++++------ services/worker/handlers.py | 43 ++- services/worker/main.py | 55 ++-- 22 files changed, 936 insertions(+), 872 deletions(-) create mode 100644 USER_GUIDE.md diff --git a/README.md b/README.md index 07d76d5..00d134c 100644 --- a/README.md +++ b/README.md @@ -80,8 +80,11 @@ uv run python -m scripts.redis_budget # projects month-end burn, exit ## Using the API -The API documents itself. FastAPI generates OpenAPI from the same models and routes it -serves, so the spec cannot drift from the implementation the way a hand-written one does. +**[USER_GUIDE.md](USER_GUIDE.md)** is the guide for callers: auth, the catalog, every +endpoint with curl, the lifecycle, the error table, the CLI. + +The API also documents itself — FastAPI generates OpenAPI from the same models and routes +it serves, so the spec cannot drift the way a hand-written one does. | What | Where | |---|---| @@ -90,24 +93,6 @@ serves, so the spec cannot drift from the implementation the way a hand-written | Raw spec, for generating clients | `https://svcforge.oci-oci.duckdns.org/openapi.json` | Locally, `uv run uvicorn services.api.main:app --factory` then . - -Three things a caller needs that a schema cannot state on its own, so they are written into -the spec's description and rendered at the top of `/docs`: - -- **Every write is asynchronous.** `POST` and `DELETE` return `202 Accepted` and enqueue - work. Poll `GET /v1/instances/{id}` and watch `state`; only `ready` carries an endpoint. -- **Authorisation is a WHERE clause.** Another team's instance returns `404`, not `403`, so - the API never confirms that an id you cannot access exists. -- **Every non-2xx body is `{"code", "message"}`**, including the 404s and 405s raised by - the framework itself, so clients never branch on the body's shape. - -Generate a client from the spec rather than hand-rolling one: - -```bash -curl -s https://svcforge.oci-oci.duckdns.org/openapi.json > openapi.json -# e.g. openapi-generator-cli generate -i openapi.json -g python -o ./client -``` - `tests/integration/test_api.py` pins the description, the tags and the bearer security scheme, so the docs fail CI if they rot. diff --git a/USER_GUIDE.md b/USER_GUIDE.md new file mode 100644 index 0000000..426e788 --- /dev/null +++ b/USER_GUIDE.md @@ -0,0 +1,267 @@ +# svcforge — user guide + +Ask for a managed service, get one. svcforge provisions Elasticsearch, Redis, Postgres and +a couple of tiny test services into Kubernetes, one namespace per team, and keeps them +matching what the database says they should be. + +**Base URL:** `https://svcforge.oci-oci.duckdns.org` + +| | | +|---|---| +| Swagger UI (send requests from the browser) | [`/docs`](https://svcforge.oci-oci.duckdns.org/docs) | +| ReDoc (nicer to read) | [`/redoc`](https://svcforge.oci-oci.duckdns.org/redoc) | +| Raw OpenAPI spec | [`/openapi.json`](https://svcforge.oci-oci.duckdns.org/openapi.json) | + +The spec is generated from the same models and routes the server runs, so it cannot drift +from the implementation. Generate a client from it rather than hand-rolling one. + +--- + +## The three things that will surprise you + +1. **Writes are asynchronous.** `POST` and `DELETE` return **202 Accepted**. Nothing is + provisioned when you get the response — you have a row and a queued task. Poll + `GET /v1/instances/{id}` and watch `state`. +2. **Authorisation is a WHERE clause.** Another team's instance returns **404**, not 403. + The API never confirms that an id you cannot access exists. +3. **Every non-2xx body is `{"code", "message"}`** — including the 404s and 405s raised by + the framework itself. Never branch on the body's shape. + +--- + +## Getting a token + +Every `/v1` route needs `Authorization: Bearer `. The token is verified against the +configured JWKS (RS256 only), and its **`team` claim** decides which instances you see. +`aud`, `iss` and `exp` are all required and all checked. + +> **The public deployment currently issues no tokens.** `SVCFORGE_JWKS_URL` points at +> `https://auth.oci-oci.duckdns.org/realms/svcforge/...`, and no identity provider is +> deployed there — the hostname resolves to the ingress, which answers with its default +> self-signed certificate. The API logs `JWKS warm-up failed` at startup and every `/v1` +> request returns `401`. Unauthenticated routes (`/healthz`, `/readyz`, `/metrics`, +> `/docs`, `/openapi.json`) work normally. To make the live API usable, either deploy an +> OIDC provider at that realm URL or repoint `auth.jwksUrl` / `auth.issuer` in +> `deploy/chart/values.yaml` at one that exists. + +Any OIDC provider works. With Keycloak, the client-credentials flow is: + +```bash +TOKEN=$(curl -s -X POST \ + https://auth.example.com/realms/svcforge/protocol/openid-connect/token \ + -d grant_type=client_credentials \ + -d client_id=svcforge-cli \ + -d client_secret="$CLIENT_SECRET" | jq -r .access_token) +``` + +The provider must put a `team` claim in the token (a Keycloak protocol mapper, or the +equivalent) and set `aud: svcforge`. A token without a non-empty string `team` is a 401. + +### Running it locally instead + +The fastest way to actually drive the API is to run it yourself with auth off: + +```bash +export SVCFORGE_PG_DSN="postgresql://svcforge:svcforge@127.0.0.1:5432/svcforge" +export SVCFORGE_AUTH_DISABLED=true # refused unless SVCFORGE_ENVIRONMENT=local +uv run uvicorn services.api.main:app --factory +``` + +Every request is then team `platform` and no header is needed. `check_production()` refuses +this flag whenever `SVCFORGE_ENVIRONMENT` is anything but `local`, so it cannot escape a +laptop. + +--- + +## The catalog + +`service_type` and `size` must both exist in the catalog. An unknown `service_type` is a +**404**; a real service type with an unknown size is a **422** that lists the valid sizes. + +| `service_type` | `size` | Memory request | Notes | +|---|---|---|---| +| `elasticsearch` | `small`, `medium` | 1Gi / 4Gi per replica | 1 or 3 replicas | +| `redis` | `small`, `medium` | 256Mi / 1Gi | 1 or 3 replicas | +| `postgres` | `small`, `medium` | 512Mi / 2Gi | 1 or 2 replicas | +| `podinfo` | `small`, `medium` | 16Mi / 32Mi | tiny, for exercising the platform | +| `nginx` | `small`, `medium` | 32Mi / 64Mi | tiny, for exercising the platform | + +Use `podinfo` or `nginx` to exercise the control plane: they are single small pods and fit +on a node with no room for a real Elasticsearch. Both are pulled straight from an OCI +registry. The three larger entries reference a `bitnamilegacy/` chart repo that the worker +image does not currently configure, so they will fail at provision time until it is added. + +--- + +## Endpoints + +### `POST /v1/instances` → 202 + +```bash +curl -X POST https://svcforge.oci-oci.duckdns.org/v1/instances \ + -H "Authorization: Bearer $TOKEN" \ + -H 'Content-Type: application/json' \ + -d '{"service_type": "podinfo", "size": "small", "ttl_days": 7}' +``` + +```json +{ + "id": "0f8b7d3e-1c2a-4f5b-9e6d-7a8b9c0d1e2f", + "state": "requested", + "service_type": "podinfo", + "size": "small", + "endpoint": null, + "chart_version": "6.7.1", + "error": null +} +``` + +The response carries a `Location` header pointing at the instance. `ttl_days` (1–30, +optional) deletes the instance automatically; omit it for no expiry. Unknown body fields +are rejected with a 422 rather than ignored. + +### `GET /v1/instances/{id}` → 200 + +The polling endpoint. Repeat until `state` is `ready` or `failed`. + +```bash +curl -H "Authorization: Bearer $TOKEN" \ + https://svcforge.oci-oci.duckdns.org/v1/instances/$ID +``` + +### `GET /v1/instances` → 200 + +Your team's instances, newest first. `?limit=` accepts 1–200 and defaults to 50. No +endpoint here returns "all rows". + +### `DELETE /v1/instances/{id}` → 202 + +```bash +curl -X DELETE -H "Authorization: Bearer $TOKEN" \ + https://svcforge.oci-oci.duckdns.org/v1/instances/$ID +``` + +Moves the instance to `deleting` and queues the teardown; the helm uninstall has not +happened when this returns. Deleting something already `deleting` or `deleted` is a **409**. + +### Unauthenticated + +`GET /healthz` (liveness, no I/O) · `GET /readyz` (readiness, checks Postgres only) · +`GET /metrics` (Prometheus exposition). + +--- + +## The lifecycle + +``` +requested ──> provisioning ──> ready ──> deleting ──> deleted + │ │ + └──────────────┴──> failed +``` + +| State | What it means | +|---|---| +| `requested` | The row exists and a provision task is queued. | +| `provisioning` | A worker is running `helm upgrade --install`. | +| `ready` | The release is up. **Only this state carries a usable `endpoint`.** | +| `failed` | The provision exhausted its retries. `error` says why. | +| `deleting` | Teardown queued or running. | +| `deleted` | Terminal. | + +`endpoint` is in-cluster DNS — +`http://.tenant-.svc.cluster.local` — reachable from inside the cluster, not +from your laptop. + +A few things happen without you asking: + +- **Drift repair.** If a `ready` instance's helm release disappears, the reconciler notices + within ~60s and re-provisions it. You may see `ready → failed → provisioning → ready`. +- **TTL.** An instance past `expires_at` is torn down automatically. +- **Upgrades.** When the catalog pins a newer chart version, instances are upgraded inside + their maintenance window, one at a time. Entries marked `security: true` skip the window. + +A `failed` instance is not retried automatically — it needs a human. + +--- + +## Errors + +Every non-2xx body is the same shape. `code` is stable and meant for machines; `message` is +for humans and must not be parsed. + +```json +{"code": "unknown_service_type", "message": "no such service_type: mongodb"} +``` + +| Status | `code` | Cause | +|---|---|---| +| 401 | `unauthorized` | Missing, expired, malformed or unverifiable token. Never says which. | +| 404 | `unknown_service_type` | Not in the catalog. | +| 404 | `not_found` | No such instance — **or it belongs to another team**. | +| 409 | `illegal_transition` | e.g. deleting something already deleted. | +| 409 | `conflict` | The row changed between the read and the write. Retry. | +| 422 | `unknown_size` | The message lists the sizes that exist. | +| 422 | `validation_error` | Malformed body: bad type, extra field, `ttl_days` out of range. | +| 429 | `rate_limited` | Over the per-team budget. Honour `Retry-After`. | +| 503 | `not_ready` | `/readyz` only: Postgres is unreachable. | + +**Rate limiting** is per team, 60 requests/minute by default, and it fails *open* — if the +limiter's Redis is down you are unmetered rather than refused. A 429 always means a real, +counted overage. + +--- + +## The CLI + +`services/cli/` is a thin API client. It holds a URL and a token and never touches the +database. + +```bash +export SVCFORGE_API_URL=https://svcforge.oci-oci.duckdns.org +export SVCFORGE_API_TOKEN="$TOKEN" + +svcforge create podinfo --size small --ttl 7d --wait +svcforge list --state ready +svcforge status +svcforge delete --yes +``` + +`--wait` polls until the instance reaches `ready` or `failed`. + +--- + +## Generating a client + +```bash +curl -s https://svcforge.oci-oci.duckdns.org/openapi.json > openapi.json +openapi-generator-cli generate -i openapi.json -g python -o ./client +``` + +The error models are declared on every route, so a generated client gets typed 401/404/409/ +422 bodies rather than guessing. `tests/integration/test_api.py` pins the description, the +tags and the bearer security scheme, so these docs fail CI if they rot. + +--- + +## Polling, end to end + +```bash +ID=$(curl -s -X POST https://svcforge.oci-oci.duckdns.org/v1/instances \ + -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \ + -d '{"service_type":"podinfo","size":"small","ttl_days":1}' | jq -r .id) + +while :; do + BODY=$(curl -s -H "Authorization: Bearer $TOKEN" \ + https://svcforge.oci-oci.duckdns.org/v1/instances/$ID) + STATE=$(jq -r .state <<<"$BODY") + echo "$STATE" + case "$STATE" in + ready) jq -r .endpoint <<<"$BODY"; break ;; + failed) jq -r .error <<<"$BODY"; exit 1 ;; + esac + sleep 5 +done +``` + +Poll every few seconds, not every few milliseconds. A provision is a helm install against a +StatefulSet; `podinfo` takes seconds, Elasticsearch takes minutes. diff --git a/libs/svcforge_core/svcforge_core/adapters/clock.py b/libs/svcforge_core/svcforge_core/adapters/clock.py index ec855f5..50327e1 100644 --- a/libs/svcforge_core/svcforge_core/adapters/clock.py +++ b/libs/svcforge_core/svcforge_core/adapters/clock.py @@ -1,18 +1,15 @@ """Time, as a dependency. -The centrepiece of the Day-2 module, and it is nine lines. `datetime.now()` called from -inside domain logic is an untestable global read: a maintenance-window test that wants -"03:00 next Sunday" would have to either sleep until Sunday or monkeypatch a stdlib symbol -and hope nothing else in the process noticed. Passing a Clock makes the same test a -`FakeClock(start=...)` and an `advance()`. +`datetime.now()` inside domain logic is an untestable global read: a maintenance-window +test wanting "03:00 next Sunday" must sleep until Sunday or monkeypatch a stdlib symbol. +Passing a Clock makes it a `FakeClock(start=...)` and an `advance()`. -Aware UTC, always. A naive datetime is a bug that survives every test on a UTC CI box and -detonates the first time it meets a tenant in Asia/Ho_Chi_Minh: `datetime.utcnow()` returns -a naive value, and comparing it to a `timestamptz` from Postgres raises TypeError, or worse, -silently compares wrong after somebody "fixes" it with a `.replace(tzinfo=...)`. +Aware UTC, always. `datetime.utcnow()` returns a naive value that survives every test on a +UTC CI box, then raises TypeError against a `timestamptz` from Postgres — or compares wrong +after someone "fixes" it with `.replace(tzinfo=...)`. -The fake lives in `tests/fakes.py`, not here: shipping test doubles in the production -package is how they end up imported by production code. +The fake lives in `tests/fakes.py`: test doubles shipped in the production package end up +imported by production code. """ from __future__ import annotations diff --git a/libs/svcforge_core/svcforge_core/adapters/helm.py b/libs/svcforge_core/svcforge_core/adapters/helm.py index 9f1f76d..8ce2d3a 100644 --- a/libs/svcforge_core/svcforge_core/adapters/helm.py +++ b/libs/svcforge_core/svcforge_core/adapters/helm.py @@ -1,19 +1,16 @@ """Driving helm from asyncio, with timeouts that actually kill helm. -The whole module exists for `_run`. Everything above it is argv construction. - -Four things go wrong when you spawn a process from an event loop, and all four are -handled here rather than in the caller: +The module exists for `_run`; everything above it is argv construction. Four things go +wrong when an event loop spawns a process, and all four are handled here: 1. `subprocess.run` blocks the loop. Use `create_subprocess_exec`. -2. `stdout=PIPE` with `proc.wait()` and nobody draining deadlocks at ~64 KB of output — - `helm --debug` clears that in one install. Use `communicate()`. -3. `asyncio.wait_for` cancels the *coroutine*. The process does not know it was waited on: - helm keeps running and keeps mutating the cluster. The timeout has to kill it. -4. `proc.kill()` signals the direct child. `helm` forks; its children reparent to init and - survive. Only `killpg` gets the whole tree, and only if the group exists — which needs - `start_new_session=True` **at spawn time**, because setsid can only run in the window - between fork and exec. +2. `stdout=PIPE` with nobody draining deadlocks at ~64 KB — one `helm --debug` install. + Use `communicate()`. +3. `asyncio.wait_for` cancels the *coroutine*; helm keeps running and keeps mutating the + cluster. The timeout has to kill it. +4. `proc.kill()` signals the direct child, and helm's children reparent and survive. Only + `killpg` gets the tree, and only with `start_new_session=True` at spawn time — setsid + can only run between fork and exec. """ from __future__ import annotations @@ -33,41 +30,35 @@ from svcforge_core.adapters.tempyaml import yaml_tempfile from svcforge_core.domain.models import CatalogEntry from svcforge_core.errors import SvcforgeError -# How long the process group gets to honour SIGTERM before SIGKILL. Helm traps SIGTERM -# and tries to leave the release in a coherent state; give it a moment to do so. +# Grace for SIGTERM before SIGKILL. Helm traps SIGTERM and tries to leave the release +# coherent; give it a moment. _TERM_GRACE_S = 5.0 -# `_run` is the backstop, not the primary timeout: helm gets its own `--timeout` so that -# `--atomic` can roll back cleanly. `_run` only fires when helm itself is wedged, so its -# deadline sits this far past helm's. +# `_run` is the backstop, not the primary timeout: helm gets its own `--timeout` so +# `--atomic` can roll back cleanly. `_run` fires only when helm itself is wedged. _RUN_TIMEOUT_MARGIN_S = 30 _STDERR_TAIL_BYTES = 2048 -# The label every release svcforge provisions carries, and the only thing that lets the -# reconciler tell its own releases from the rest of the cluster's. Written by install(), -# read by list_releases(). Changing either value without the other silently empties the -# reconciler's view of reality, which reads as "no drift" rather than as an error. -# -# `app.kubernetes.io/managed-by` is the standard key for exactly this, so anyone reading -# the cluster with kubectl gets the same answer the reconciler does. +# The label every svcforge release carries — written by install(), read by list_releases(). +# It is the only thing that tells svcforge's releases from the rest of the cluster's. +# Changing one value without the other empties the reconciler's view, which reads as +# "no drift" rather than as an error. MANAGED_BY_LABEL = "app.kubernetes.io/managed-by" MANAGED_BY_VALUE = "svcforge" # Where the kubelet mounts the pod's ServiceAccount. Their presence is also how this module -# decides it is running inside the cluster: in-cluster gets the fast release read below, -# anything else falls back to shelling helm. +# decides it is in-cluster: in-cluster takes the fast release read, everything else shells +# out to helm. _SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount") _SA_TOKEN = _SA_DIR / "token" _SA_CA = _SA_DIR / "ca.crt" -# helm's own label on every release secret it writes. Pairing it with MANAGED_BY_LABEL is -# what separates svcforge's releases from the rest of the cluster's. +# helm's own label on every release secret it writes. _HELM_OWNER_LABEL = "owner=helm" -# The states `helm list` shows by default. Pushed into the selector so superseded revisions -# never leave the API server: this cluster had 96 release secrets of which 71 were -# superseded, so the filter is most of the win. +# The states `helm list` shows by default. In the selector so superseded revisions never +# leave the API server — 96 release secrets here, 71 of them superseded. _LIVE_STATUSES = "status in (deployed,failed,pending-install,pending-upgrade,pending-rollback)" _API_TIMEOUT_S = 15.0 @@ -77,7 +68,7 @@ class HelmError(SvcforgeError, RuntimeError): """Non-zero exit. str(self) is the stderr tail that lands in instances.error. RuntimeError stays in the MRO so callers written against it keep catching; SvcforgeError - comes first so `except SvcforgeError` can separate a modelled failure from a stray bug. + comes first so `except SvcforgeError` separates a modelled failure from a stray bug. """ @@ -107,9 +98,9 @@ class Provisioner(Protocol): def _tail(raw: bytes, tail_bytes: int) -> str: """The last `tail_bytes` of a stream, as text. - Truncation happens here, at the adapter boundary, and nowhere else. A helm failure can - emit megabytes; `instances.error` is a text column read by humans. Slice the bytes, not - the decoded string, then decode with `replace` — the cut can land mid-codepoint. + Truncation happens here, at the adapter boundary, and nowhere else: a helm failure can + emit megabytes and `instances.error` is read by humans. Slice the bytes and decode with + `replace` — the cut can land mid-codepoint. """ return raw[-tail_bytes:].decode("utf-8", errors="replace").strip() @@ -117,8 +108,8 @@ def _tail(raw: bytes, tail_bytes: int) -> str: def _signal_group(proc: asyncio.subprocess.Process, sig: int) -> None: """Signal the process's whole group. No-op if it has already exited. - `os.getpgid` rather than `proc.pid`: with `start_new_session=True` they are equal, but - that equality is an implementation detail, and asking the kernel costs nothing. + `os.getpgid` rather than `proc.pid`: `start_new_session=True` makes them equal, but that + equality is an implementation detail and asking the kernel costs nothing. """ if proc.returncode is not None: return @@ -181,8 +172,8 @@ class HelmProvisioner: self, *, helm_bin: str = "helm", kubeconfig: Path | None = None, timeout_s: int = 600 ) -> None: """kubeconfig=None means the ambient config: $KUBECONFIG, ~/.kube/config, or the - in-cluster service account when svcforge runs as a pod. Keyword-only so that the - three arguments can never be swapped by position at a call site. + in-cluster service account. Keyword-only so the three can never be swapped by + position at a call site. """ self._helm_bin = helm_bin self._kubeconfig = kubeconfig @@ -201,10 +192,10 @@ class HelmProvisioner: async def _run_helm(self, argv: Sequence[str]) -> str: """`_run`, with the timeout path translated to this adapter's declared error type. - `_run` raises a bare TimeoutError so that the process-group test can assert on it - directly, but every public method here is documented as raising HelmError; a wedged - helm arriving as TimeoutError sails straight past a caller's `except HelmError` and - fails the task as an unmodelled crash. Translate once, at the public boundary. + `_run` raises a bare TimeoutError so the process-group test can assert on it, but + every public method here is documented as raising HelmError. A wedged helm arriving + as TimeoutError sails past a caller's `except HelmError` and fails the task as an + unmodelled crash, so translate once, at the public boundary. """ try: return await _run(argv, timeout_s=self._run_timeout_s) @@ -213,11 +204,10 @@ class HelmProvisioner: async def install(self, release: str, ns: str, entry: CatalogEntry, values: dict[str, Any]) -> None: """helm upgrade --install --wait --timeout. Idempotent by construction.""" - # `upgrade --install` is why this is idempotent: a retried task after a crash mid-provision - # converges on the same release instead of erroring with "release already exists". - # `--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. + # `upgrade --install`: a task retried after a crash mid-provision converges on the + # same release instead of erroring with "release already exists". `--wait` is why + # `ready` in the DB means ready. `--atomic` rolls back a failed upgrade and doubles + # the worst case, which is what the two timeouts are sized around. with yaml_tempfile(values, prefix="svcforge-values-", name="values.yaml") as path: argv = self._base_argv( "upgrade", @@ -226,17 +216,13 @@ class HelmProvisioner: entry.chart, "--namespace", ns, - # `--namespace X` does not create X. Every tenant's first provision targets - # a namespace that does not exist yet, and helm fails with "namespaces not - # found". helm creates it here rather than a separate `kubectl apply` step, - # which keeps kubectl out of the worker image entirely — one fewer binary, - # and one fewer set of vendored Go CVEs to track. Idempotent: existing - # namespaces are left alone. + # `--namespace X` does not create X, and every tenant's first provision + # targets one that does not exist yet. helm creates it here rather than a + # `kubectl apply` step, which keeps kubectl out of the worker image — one + # fewer binary and one fewer set of vendored Go CVEs. Idempotent. "--create-namespace", - # Stamps MANAGED_BY_LABEL onto the release, which is what makes - # list_releases() able to ask for svcforge's releases and nobody else's. - # Without it the reconciler has to list every release in the cluster and - # sort out ownership afterwards, which it cannot actually do. + # Stamps MANAGED_BY_LABEL, which is what lets list_releases() ask for + # svcforge's releases and nobody else's. "--labels", f"{MANAGED_BY_LABEL}={MANAGED_BY_VALUE}", "--version", @@ -265,36 +251,24 @@ class HelmProvisioner: await self._run_helm(argv) async def list_releases(self) -> list[ReleaseInfo]: - """Every release SVCFORGE provisioned, in every namespace. The reconciler's view of reality. + """Every release svcforge provisioned, in every namespace. The reconciler's reality. - Scoped by label, and the scope is load-bearing twice over. - - Correctness first. The reconciler diffs this against the database in both directions, - and the second direction is `live - known` -> `drift.orphan_release` at ERROR. + Scoped by label for correctness: the reconciler diffs this against the database in + both directions, and `live - known` is reported as `drift.orphan_release` at ERROR. Unscoped, `live` is every release in the cluster, so argocd, longhorn, gitea and - cert-manager are all reported as orphans svcforge is failing to account for, every - sweep. They are not orphans. They were never svcforge's to know about. + cert-manager are all reported as orphans on every sweep. - Cost is why this does not shell out to helm when it does not have to. `--selector` is - 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. + In-cluster this reads the release secrets off the API server instead of shelling + out — 21ms and 31KB against helm's 4392ms, because helm applies `--selector` only + after fetching and decompressing every release secret in the cluster. That cost was + not theoretical: with the CPU request mutated to 0 by a cluster policy, helm's list + took over 330s and timed out on every tick, and a check that never completes reports + no drift. Out of cluster there is no ServiceAccount, so it falls back to helm and the + e2e suite keeps working. See `_list_releases_via_api`. - 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. - - In-cluster this reads the release secrets directly instead. Measured from a pod on - this cluster: 21ms and 31KB against helm's 4392ms, because nothing is decompressed - and no release payload crosses the wire. See `_list_releases_via_api`. Out of cluster there is no - ServiceAccount to authenticate with, so it falls back to helm — which keeps the e2e - suite, and anyone running this from a laptop, working unchanged. - - Releases provisioned before the label existed will not match, so the first sweep - after this ships sees them as missing and re-provisions. That is safe by - construction — provisioning is `helm upgrade --install` against a deterministic - release name — and the re-provision is what applies the label. + Releases provisioned before the label existed do not match, so the first sweep sees + them as missing and re-provisions. That is safe — provisioning is `upgrade --install` + against a deterministic release name — and the re-provision applies the label. """ via_api = await self._list_releases_via_api() if via_api is not None: @@ -319,51 +293,34 @@ class HelmProvisioner: async def _list_releases_via_api(self) -> list[ReleaseInfo] | None: """Release names and namespaces read straight off helm's release secrets. - Returns None when there is no in-cluster ServiceAccount to authenticate with, which - is the caller's signal to fall back to helm. + None when there is no in-cluster ServiceAccount, which is the caller's signal to + fall back to helm. - This exists because helm's own list is expensive for a reason this caller does not - need. helm gunzips every release payload to build its table; the only fields anyone - here reads are name and namespace: + helm gunzips every release payload to build its table. The only fields either caller + reads are name and namespace, and both live in the secret's labels and metadata, so + nothing has to be decompressed. Three details carry the correctness: - reconciler/main.py live = {(r.name, r.namespace) for r in releases} - worker/handlers.py releases = {r.name for r in await ...list_releases()} + * `PartialObjectMetadataList` in the Accept header asks for metadata only. Without + it the response carries every release's gzipped manifest — megabytes fetched to be + thrown away, which is the cost this method exists to avoid. + * The status selector drops superseded revisions server-side (96 secrets here, 25 + live). The states kept are the ones `helm list` shows, so a failed release still + counts as existing — it does, and calling it missing would re-provision on top. + * helm writes one secret per revision, so a release can appear several times. The + newest `version` label wins; without that, any caller counting releases over-counts. - Both live in the secret's labels and its metadata, so nothing has to be decompressed. - That is the whole difference between ~141ms and ~4.4s, and it is why this survives a - container whose CPU request was mutated to 0. - - Three details carry the correctness: - - * `PartialObjectMetadataList` in the Accept header asks the API server for metadata - only. Without it the response carries every release's gzipped manifest — megabytes - of payload fetched purely to be thrown away, which is the cost this method exists - to avoid. - * The status selector drops superseded revisions server-side: 96 release secrets on - this cluster, 25 of them live. The states kept are the ones `helm list` shows by - default, so a failed release still counts as existing — it does exist, and - 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 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` 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. + `chart` comes back empty because the chart name lives only in the compressed payload. + The field stays so the helm fallback, which does populate it, 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. + # The CA is checked here rather than left to httpx, which loads it eagerly at client + # construction and raises OSError — not in the except below. A half-mounted + # ServiceAccount would crash the tick as a bare bug instead of falling back. A + # complete ServiceAccount is the in-cluster signal; a missing CA means "not + # in-cluster" exactly as a missing token does. if not token or not os.access(_SA_CA, os.R_OK): return None host, port = ( @@ -376,11 +333,10 @@ 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 + # No `limit`, and that is load-bearing: the apiserver only returns a + # `metadata.continue` token when the client sets one, so the single read + # below is the complete set. Adding `limit` without looping on `continue` + # would truncate silently, 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", @@ -393,17 +349,14 @@ class HelmProvisioner: resp.raise_for_status() # `or []`, not `.get("items", [])`. Kubernetes serialises an empty list as # `"items": null`, so the key is present and the default never fires. This - # shipped and failed in production on the first tick that matched no - # releases: TypeError: 'NoneType' object is not iterable. + # shipped and failed on the first tick that matched no releases. items: Any = resp.json().get("items") or [] except (httpx.HTTPError, json.JSONDecodeError) as exc: # Raise, do not fall back to helm. The fallback is for "there is no - # ServiceAccount here", which is a fact about the environment and is known - # before any request goes out. This is different: the API server was reachable - # and something went wrong, and quietly retrying through helm would swap a - # visible error for the 330s timeout this method exists to remove, on a - # container that OOMs while helm parses. The tick logs check.failed and the - # next one tries again in 60s. + # ServiceAccount", a fact about the environment known before any request goes + # out. Here the API server was reachable and something went wrong, and retrying + # through helm would swap a visible error for the 330s timeout this method + # exists to remove. The tick logs check.failed and tries again in 60s. raise HelmError(f"listing release secrets failed: {exc}") from exc newest: dict[tuple[str, str], tuple[int, ReleaseInfo]] = {} diff --git a/libs/svcforge_core/svcforge_core/adapters/notify.py b/libs/svcforge_core/svcforge_core/adapters/notify.py index 78287fe..1ada443 100644 --- a/libs/svcforge_core/svcforge_core/adapters/notify.py +++ b/libs/svcforge_core/svcforge_core/adapters/notify.py @@ -1,12 +1,11 @@ """Telling someone a provision finished, or didn't. -Best-effort by construction: `send` never raises. A notifier that can fail a task is a -notifier that lets a Slack outage roll back a successful provision. The instance is ready; -the DB says so; failing the task would re-run helm for nothing. Delivery failures are -logged and dropped on the floor, which is the correct amount of ceremony for a webhook. +Best-effort by construction: `send` never raises. A notifier that can fail a task lets a +Slack outage roll back a successful provision — the instance is ready and the DB says so, +and failing the task would re-run helm for nothing. Delivery failures are logged and dropped. -Two implementations, so the Protocol earns its place: LogNotifier (the default, and what -tests and local dev get) and WebhookNotifier (the one that leaves the process). +Two implementations, so the Protocol earns its place: LogNotifier (the default) and +WebhookNotifier (the one that leaves the process). """ from __future__ import annotations @@ -47,18 +46,14 @@ class LogNotifier: """Writes the event to the log. The default: structured logs are already shipped somewhere.""" async def send(self, event: str, message: str, fields: dict[str, str] | None = None) -> None: - # structlog kwargs, NOT logging's `extra=`. obs bridges stdlib records through - # ProcessorFormatter, which builds the event dict from `record.msg` alone — every - # key passed via `extra=` is dropped on the floor, so the default notifier used to - # emit a bare {"event": "notify"} with the payload gone. + # structlog kwargs, NOT logging's `extra=`: ProcessorFormatter builds the event dict + # from `record.msg` alone, so `extra=` keys are dropped and this used to emit a bare + # {"event": "notify"} with the payload gone. # - # Fields are splatted rather than nested under "fields" so each one is its own - # queryable key in Loki. `detail`, not `message`: `message` is a reserved LogRecord - # attribute and the stdlib bridge raises KeyError on it. - # - # `notify_event`, not `event`: structlog's first positional parameter IS named - # `event` (it becomes the rendered line's "event" key, here the literal "notify"), - # so passing event= alongside it is a TypeError at the call, not a rename. + # Fields are splatted rather than nested so each is its own queryable key in Loki. + # `detail`, not `message` — a reserved LogRecord attribute the bridge raises on. And + # `notify_event`, not `event` — structlog's first positional parameter is named + # `event`, so passing it as a kwarg is a TypeError at the call. log = obs.get_logger(__name__) log.info("notify", notify_event=event, detail=message, **_safe_fields(fields)) @@ -77,8 +72,8 @@ class WebhookNotifier: self._timeout_s = timeout_s self._owns_client = client is None # Eager, not lazy. `AsyncClient()` does no I/O, so laziness bought nothing and cost a - # race: two concurrent `send`s could both see None, both construct a client, and the - # loser's connection pool would leak because only one of them survived the assignment. + # race: two concurrent `send`s both see None, both construct a client, and the loser's + # connection pool leaks because only one survives the assignment. self._client = client if client is not None else httpx.AsyncClient(timeout=self._timeout_s) async def send(self, event: str, message: str, fields: dict[str, str] | None = None) -> None: @@ -87,10 +82,9 @@ class WebhookNotifier: resp = await self._client.post(self._url, json=payload, timeout=self._timeout_s) resp.raise_for_status() except Exception as exc: # the bare `except Exception` IS the specification here - # `send` must not raise; that is the contract in the module docstring, and it is - # not satisfiable by catching httpx.HTTPError alone. `httpx.InvalidURL` is not an - # HTTPError subclass, and posting on an already-aclose()d client raises - # RuntimeError — so a typo'd webhook URL would fail a task whose helm work has + # "send must not raise" is not satisfiable by catching httpx.HTTPError alone: + # `httpx.InvalidURL` is not a subclass, and posting on an aclose()d client raises + # RuntimeError — so a typo'd webhook URL would fail a task whose helm work # already succeeded. exc_info so the traceback survives the swallowing. obs.get_logger(__name__).warning( "notify webhook failed", notify_event=event, error=str(exc), exc_info=exc @@ -99,9 +93,9 @@ class WebhookNotifier: async def aclose(self) -> None: """Close the client, if we made it. Call at process shutdown, next to the pool's close. - The client reference is kept rather than cleared: a `send` that races shutdown now - raises RuntimeError on a closed client, and `send` swallows and logs that like any - other delivery failure instead of resurrecting a pool nobody will close. + The reference is kept rather than cleared: a `send` racing shutdown then raises + RuntimeError on a closed client and is swallowed like any other delivery failure, + instead of resurrecting a pool nobody will close. """ if self._owns_client: await self._client.aclose() diff --git a/libs/svcforge_core/svcforge_core/adapters/redis.py b/libs/svcforge_core/svcforge_core/adapters/redis.py index ebf7035..cc7b028 100644 --- a/libs/svcforge_core/svcforge_core/adapters/redis.py +++ b/libs/svcforge_core/svcforge_core/adapters/redis.py @@ -1,39 +1,29 @@ """Redis: derived state only. Never the truth, never the queue. -Everything in here is a shortcut past Postgres, and every one of them is optional. Postgres -holds the instances, the tasks, the leases and the `release_name` UNIQUE constraint. Redis -holds a counter, a claim marker and a copy — all of it rebuildable by doing nothing and -waiting for a TTL. +Everything here is an optional shortcut past Postgres, which holds the instances, the +tasks, the leases and the `release_name` UNIQUE constraint. Redis holds a counter, a claim +marker and a copy — all rebuildable by waiting for a TTL. -That framing decides the error handling, and the error handling is the module. Each class -below catches `RedisError` and returns a *safe* answer rather than raising: +That decides the error handling, and the error handling is the module. Each class catches +`RedisError` and returns a safe answer rather than raising: -| Path | Redis is down | Why | -|-------------|--------------------------|--------------------------------------------------| -| Cache | miss -> read Postgres | It was an optimisation. Nobody notices. | -| Rate limit | **allow** | An internal platform that refuses every request | -| | | because the limiter is sick is worse than one | -| | | that is briefly unmetered. | -| Idempotency | fall through to the DB | `instances.release_name` is UNIQUE. That is the | -| | | real guarantee; this is the fast path. | +| Path | Redis is down | Why | +|-------------|------------------------|--------------------------------------------------| +| Cache | miss -> read Postgres | It was an optimisation. Nobody notices. | +| Rate limit | **allow** | Briefly unmetered beats refusing every request. | +| Idempotency | fall through to the DB | `instances.release_name` UNIQUE is the guarantee.| -Consequently nothing here raises out to a caller, and `/readyz` stays Postgres-only. A -Redis outage must not make a single pod unready — that would convert "the cache is down" -into "the platform is down", which is the exact inversion this module exists to prevent. +Nothing here raises out to a caller and `/readyz` stays Postgres-only, so a Redis outage +never makes a pod unready. -**The budget is a design constraint.** Upstash free tier: +**The budget is a design constraint.** Upstash free tier is 500,000 commands/month = +16,129/day = 0.19/second sustained. One worker polling every five seconds spends the entire +budget producing nothing, so the rule is structural: Redis lives on the request path only, +never in a poll or control loop. It is also why the limiter is a Lua script — +`GET`/`INCR`/`EXPIRE` is three billed commands and a race, one `EVALSHA` is one and atomic. +A pipeline batches round trips but still bills N. - 500,000 commands / month = 16,129 / day = 11 / minute = 0.19 / second, sustained - -One worker polling Redis every five seconds spends 518,400/month: the entire budget, -producing nothing. So the rule is structural — **Redis lives on the request path only**, -where volume is bounded by the number of humans with an API token, and never inside a poll -or control loop. That is also why the limiter is a Lua script: `GET`/`INCR`/`EXPIRE` is -three billed commands and a race; one `EVALSHA` is one billed command and atomic. A -pipeline would not help — it batches round trips but still bills N. - -Every key gets a TTL. 256 MB with no expiry is a slow leak that ends by evicting the keys -you cared about. +Every key gets a TTL. 256 MB with no expiry is a leak that ends by evicting what mattered. """ from __future__ import annotations @@ -59,22 +49,18 @@ if TYPE_CHECKING: from svcforge_core.settings import Settings -# 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. +# structlog via obs, not stdlib logging: the stdlib bridge builds the event dict from the +# record message alone and drops `extra=`. A bound logger takes fields as kwargs and keeps +# them. Bound per instance in __init__, after obs.setup() runs, never at import time. # --- The budget metric ------------------------------------------------------------------ # -# The counter that `scripts/redis_budget.py` projects month-end burn from. It counts -# commands we *send*, incremented next to each call, because the number that matters is -# the one Upstash bills — not the number of times a method was called. A cache miss calls -# `get()` once and spends one command; a `put()` after it spends another. +# What `scripts/redis_budget.py` projects month-end burn from. It counts commands *sent*, +# incremented next to each call, because the billed number is what matters — a cache miss +# spends one command on `get()` and another on the `put()` after it. # -# The dangerous failure this makes visible: when Redis is down, every path degrades -# silently and correctly, so nothing pages. Nothing fails until the month rolls over and -# every Redis call starts erroring at once. A counter you can extrapolate from is the only -# warning you get. +# This is the only warning available: when Redis is down every path degrades silently and +# correctly, so nothing pages until the month rolls over and every call starts erroring. REDIS_COMMANDS = Counter( "svcforge_redis_commands_total", @@ -88,26 +74,24 @@ REDIS_ERRORS = Counter( ["op"], ) -# A hung Redis must not hang the request path. Without these, a TCP connection that is -# open but unanswered blocks the handler until the client gives up — which turns "Redis is -# slow" into "the API is down", the same inversion the fail-open policy prevents. Upstash -# steady-state RTT is ~2.4 ms; two seconds is already pathological. +# A hung Redis must not hang the request path: an open but unanswered TCP connection blocks +# the handler until the client gives up, turning "Redis is slow" into "the API is down". +# Upstash steady-state RTT is ~2.4 ms, so two seconds is already pathological. _SOCKET_TIMEOUT_S = 2.0 _CONNECT_TIMEOUT_S = 2.0 -# Errors that mean "Redis did not answer". Every public method below turns these into a -# safe default. `OSError` because a DNS failure at connect time need not arrive wrapped, -# `TimeoutError` because the socket timeouts above raise it. +# "Redis did not answer" — every public method turns these into a safe default. `OSError` +# because a DNS failure at connect time need not arrive wrapped, `TimeoutError` because the +# socket timeouts above raise it. _REDIS_DOWN = (RedisError, OSError, asyncio.TimeoutError) def _as_text(value: bytes | str) -> str: """`decode_responses=True` already did this; redis-py's annotations do not know it. - The client is configured for text, so the `bytes` branch is unreachable in this - process. It stays because the type says it is reachable, and a `cast` here would hide - the day someone builds a client without `decode_responses` and gets a `UUID(b'...')` - TypeError from three frames away instead of a value that just works. + The `bytes` branch is unreachable in this process. It stays rather than becoming a + `cast` so that a client built without `decode_responses` gets a working value instead + of a `UUID(b'...')` TypeError three frames away. """ return value.decode() if isinstance(value, bytes) else value @@ -115,21 +99,14 @@ def _as_text(value: bytes | str) -> str: def make_redis(settings: Settings) -> Redis | None: """One client per process, opened in lifespan next to the psycopg pool, closed on exit. - `None` when no DSN is configured, and that is a supported way to run: every consumer - below is optional by construction, so "no Redis" and "Redis is down" take the same - code path. The signature is `Redis | None` rather than `Redis` precisely so that - "unconfigured" cannot be faked with a client pointed at nothing. + `None` when no DSN is configured, which is a supported way to run: every consumer is + optional, so "no Redis" and "Redis is down" take the same path. The return type is + `Redis | None` so "unconfigured" cannot be faked with a client pointed at nothing. - Two settings are not negotiable: - - `decode_responses=True` — the first bug everyone hits. Without it every read is - `bytes` and the traceback is `AttributeError: 'bytes' object has no attribute - 'encode'`, several frames away from the cause. - - `rediss://` (TLS) — Upstash rejects plaintext. The handshake is ~56 ms against a - ~2.4 ms steady-state RTT, which is the whole argument for one pooled client per - process: a client per request pays the handshake every time and turns a cache into a - latency regression. + `decode_responses=True` is not optional — without it every read is `bytes` and the + traceback is an `AttributeError` several frames from the cause. Neither is `rediss://`: + Upstash rejects plaintext, and the ~56 ms handshake against a ~2.4 ms steady-state RTT + is the whole argument for one pooled client per process. """ if settings.redis_dsn is None: return None @@ -143,15 +120,14 @@ def make_redis(settings: Settings) -> Redis | None: # --- Rate limiting ---------------------------------------------------------------------- -# One INCR; EXPIRE only when the counter is new. The `== 1` test is the entire trick: set -# the TTL unconditionally and every request slides the window forward, so a caller at -# steady load is never reset and the "window" is a sliding refusal that never lets up. +# One INCR; EXPIRE only when the counter is new. The `== 1` test is the trick: set the TTL +# unconditionally and every request slides the window forward, so a caller at steady load is +# never reset and the window becomes a refusal that never lets up. # -# KEYS and ARGV arrive as 1-based tables — Lua indexes from 1, and `ARGV[0]` is silently -# nil rather than an error, which reads as "the limit is nil" and compares false forever. +# KEYS and ARGV are 1-based — `ARGV[0]` is silently nil rather than an error, which reads as +# "the limit is nil" and compares false forever. # -# Everything derivable in Python is derived in Python: `reset_at` comes from the window -# number the caller already computed, so there is no TTL round trip. One command, total. +# `reset_at` is derived in Python from the window number, so there is no TTL round trip. _RATE_LIMIT_LUA = """ local n = redis.call('INCR', KEYS[1]) if n == 1 then @@ -178,9 +154,9 @@ 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. + # From the same injected clock as reset_at. The two must share a clock or retry_after_s + # (their difference) is meaningless under a FakeClock and drifts by the request latency + # in production. checked_at: datetime degraded: bool = False @@ -206,12 +182,9 @@ class RateLimiterProto(Protocol): class RateLimiter: """Fixed-window limiter. One EVALSHA per check. Fails OPEN. - Fixed window, not a token bucket or a sliding log, because the window boundary is the - only thing a fixed window gets wrong and the cost of getting it wrong is that a caller - can spend 2x the limit across a boundary. A sliding log is a sorted set, an - `ZREMRANGEBYSCORE`, an `ZADD` and a `ZCARD` — four billed commands and unbounded key - size — to fix a burst nobody is paying for. The limit is a courtesy, not a security - control; the security control is the JWT. + A fixed window's only error is the boundary, where a caller can spend 2x the limit. The + sliding log that fixes it costs four billed commands and an unbounded key. The limit is + a courtesy; the security control is the JWT. """ def __init__(self, r: Redis, limit: int, window_s: int, *, clock: Clock | None = None) -> None: @@ -224,10 +197,9 @@ class RateLimiter: self._limit = limit self._window_s = window_s self._clock = clock or SystemClock() - # register_script() is local: it hashes the source and returns a callable. No round - # trip here, and none wasted at import. The first call sends EVALSHA; redis-py - # catches NOSCRIPT and replays it as EVAL, which is why a restarted Redis costs one - # extra command once rather than an outage. + # register_script() is local — it hashes the source and returns a callable, with no + # round trip. The first call sends EVALSHA; redis-py catches NOSCRIPT and replays it + # as EVAL, so a restarted Redis costs one extra command rather than an outage. self._script: AsyncScript = r.register_script(_RATE_LIMIT_LUA) def _window(self) -> tuple[int, datetime]: @@ -241,9 +213,8 @@ class RateLimiter: async def check(self, team: str) -> RateLimitResult: """Count one request against `team`. Never raises. - On any Redis error: allow, log loudly, count it. The metric is the point — a - limiter that fails open silently is indistinguishable from no limiter at all, and - you find out which one you shipped during the incident. + On any Redis error: allow, log loudly, count it. The metric is the point — a limiter + that fails open silently is indistinguishable from no limiter at all. """ window, reset_at = self._window() key = f"rl:{team}:{window}" @@ -288,18 +259,15 @@ class IdempotencyStoreProto(Protocol): class IdempotencyStore: """`SET NX EX`. Maps an `Idempotency-Key` to the instance UUID it created. - Claimed BEFORE the DB transaction, so the marker exists before the row it names. The - inversion matters: claim after the commit and a crash in between leaves a created - instance with no marker, and the client's retry creates a second one. + Claimed BEFORE the DB transaction: claim after the commit and a crash in between leaves + a created instance with no marker, so the client's retry creates a second one. Claiming + first has the opposite hole — a marker naming an instance that never committed, and the + retry is told "already done" about nothing — and that is the better hole, because the + client polls the id, gets a 404 and retries with a fresh key. - Claiming first has its own hole — a crash after the claim and before the commit leaves - a marker pointing at an instance that never existed, and the retry is told "already - done" about nothing. It is the better hole: the client polls the id, gets a 404, and - retries with a fresh key. The alternative loses money to a duplicate Elasticsearch. - - And neither hole is load-bearing, because `instances.release_name` is UNIQUE and - deterministic from (team, service_type, id). **That constraint is the guarantee.** This - class only saves the round trip to find out. + Neither hole is load-bearing. `instances.release_name` is UNIQUE and deterministic from + (team, service_type, id); that constraint is the guarantee and this only saves a round + trip. """ def __init__(self, r: Redis, ttl_s: int = 86400) -> None: @@ -313,13 +281,11 @@ class IdempotencyStore: async def claim(self, key: str, instance_id: UUID) -> UUID | None: """Try to bind `key` to `instance_id`. Never raises. - `None` from the happy path means "you won, go create it". `None` from a Redis - failure means the same thing — the caller creates, and the UNIQUE constraint - catches an actual duplicate. Degrading to "create it" is safe *only* because that - constraint exists; without it this would have to fail closed. + `None` means "you won, go create it", and a Redis failure returns the same thing — + the caller creates and the UNIQUE constraint catches a real duplicate. Degrading to + "create it" is safe only because that constraint exists. - One command when we win, which is the common case and the one the budget is sized - for. Two when we lose: the loser pays a GET, and losers are rare by definition. + One command when we win, two when we lose: the loser pays a GET, and losers are rare. """ redis_key = f"idem:{key}" try: @@ -338,8 +304,8 @@ class IdempotencyStore: return None if existing is None: - # The key expired between the SET and the GET. Vanishingly rare, and the honest - # answer is "no winner recorded" — let the caller create and let Postgres decide. + # The key expired between the SET and the GET. The honest answer is "no winner + # recorded" — let the caller create and let Postgres decide. return None try: return UUID(_as_text(existing)) @@ -370,15 +336,13 @@ class InstanceCacheProto(Protocol): class InstanceCache: """Cache-aside for `GET /v1/instances/{id}`. TTL 30s. - Hit costs one command, miss costs two (the GET, then the SET after Postgres answers). - That is ~1 per read at any useful hit rate, which is what keeps a read-heavy poller - inside the budget. + A hit costs one command and a miss two, so ~1 per read at any useful hit rate — which is + what keeps a read-heavy poller inside the budget. - The TTL is short on purpose and is the actual correctness argument. `invalidate()` on - every state transition is the fast path, not the guarantee: the worker can crash - between the UPDATE and the DEL, and then the cache is wrong. Thirty seconds bounds how - wrong. Trusting the invalidation instead — and raising the TTL to an hour — is how a - deleted instance stays `ready` in the API for an hour. + The short TTL is the correctness argument. `invalidate()` on every state transition is + the fast path, not the guarantee: a worker can crash between the UPDATE and the DEL, and + 30 seconds bounds how wrong the cache gets. Trusting invalidation and raising the TTL to + an hour is how a deleted instance stays `ready` in the API for an hour. """ def __init__(self, r: Redis, ttl_s: int = 30) -> None: @@ -395,8 +359,8 @@ class InstanceCache: async def get(self, instance_id: UUID) -> Instance | None: """One GET. A miss, a Redis outage and a corrupt entry are all the same answer. - Which is the point: the caller writes `cache.get() or repo.get()` and has no branch - for "Redis is broken", because there is nothing different to do about it. + The caller writes `cache.get() or repo.get()` and has no branch for "Redis is + broken", because there is nothing different to do about it. """ try: REDIS_COMMANDS.labels(op="cache_get").inc() @@ -410,8 +374,8 @@ class InstanceCache: try: return Instance.model_validate_json(raw) 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. + # A model change deployed over a warm cache. A miss, not an error — the TTL + # takes the old shape out and the truth is in Postgres either way. self._log.info("cache entry failed validation; treating as a miss") return None @@ -427,8 +391,8 @@ class InstanceCache: async def invalidate(self, instance_id: UUID) -> None: """One DEL. Called by the worker inside the code path that writes the state. - Inside that path, not after it and not from a subscriber: an invalidation that can - be skipped by an early return is an invalidation that will be. + Inside that path, not after it and not from a subscriber: an invalidation an early + return can skip is an invalidation that will be skipped. """ try: REDIS_COMMANDS.labels(op="cache_del").inc() diff --git a/libs/svcforge_core/svcforge_core/errors.py b/libs/svcforge_core/svcforge_core/errors.py index aa1e3a6..18e5b45 100644 --- a/libs/svcforge_core/svcforge_core/errors.py +++ b/libs/svcforge_core/svcforge_core/errors.py @@ -1,13 +1,11 @@ """The one base class every svcforge-raised exception shares. -Without it, a caller that wants "the cluster failed" has to write `except Exception`, which -also swallows the `AttributeError` from a typo three frames down. The two are not the same -incident: one is retried, the other is a bug that must reach the dead-letter loudly. A single -root makes that distinction expressible in one clause. +Without it, "the cluster failed" has to be caught as `except Exception`, which also swallows +the `AttributeError` from a typo three frames down. One is retried and the other is a bug +that must dead-letter loudly; a single root makes that expressible in one clause. -Subclasses keep their existing stdlib base as well (`HelmError(SvcforgeError, RuntimeError)`), -so code already written against `except RuntimeError` keeps working. The MRO order matters: -`SvcforgeError` first, so the svcforge-specific class is the more derived one. +Subclasses keep their stdlib base too (`HelmError(SvcforgeError, RuntimeError)`), so code +written against `except RuntimeError` keeps working. `SvcforgeError` comes first in the MRO. """ from __future__ import annotations diff --git a/libs/svcforge_core/svcforge_core/obs.py b/libs/svcforge_core/svcforge_core/obs.py index ada14d3..a38b47c 100644 --- a/libs/svcforge_core/svcforge_core/obs.py +++ b/libs/svcforge_core/svcforge_core/obs.py @@ -1,33 +1,27 @@ """Logs, traces, metrics. One setup() call, made once, before anything else. -Three libraries, one module, because the three are one decision. A log line without the -trace id it belongs to is a log line you cannot join to anything; a span without the -`instance_id` the request is about is a span you cannot search for. They are wired here -together so that no service can configure two of the three and ship. +Three libraries in one module because they are one decision: a log line without its trace +id joins to nothing, and a span without the `instance_id` cannot be searched for. Wiring +them together here stops a service configuring two of the three and shipping. -The three things that make this module worth reading: +Three things worth knowing: -1. **Context does not cross a queue.** A trace is a chain of parent/child span contexts - passed in-process or over a wire header. `POST /v1/instances` inserts a row and - returns; the worker picks that row up ninety seconds later in a different pod. Nothing - carries the context across — unless we carry it ourselves. So: `inject_traceparent()` - at enqueue, a `traceparent` column, `context_from_traceparent()` at claim. Two - disconnected traces in Tempo is the symptom of skipping this. +1. **Context does not cross a queue.** `POST /v1/instances` inserts a row and returns; the + worker picks it up ninety seconds later in another pod, with no ambient context. So: + `inject_traceparent()` at enqueue, a `traceparent` column, `context_from_traceparent()` + at claim. Two disconnected traces in Tempo is the symptom of skipping this. -2. **Histogram buckets are a domain decision.** prometheus_client's defaults top out at - 10 seconds because they were chosen for HTTP handlers. A provision is `helm --wait` on - a StatefulSet: minutes. With the defaults every observation lands in `+Inf`, - `histogram_quantile` interpolates inside a bucket that spans 10s→infinity, and the p95 - it prints is a number with no relationship to reality. The buckets below are sized for - what is being measured. +2. **Histogram buckets are a domain decision.** prometheus_client's defaults were chosen + for HTTP handlers and top out at 10s; a provision is `helm --wait` on a StatefulSet, so + every observation lands in `+Inf` and the p95 is interpolated inside a bucket spanning + 10s→infinity. The buckets below are sized for what is measured. -3. **One process per pod.** prometheus_client keeps its registry in process memory. Run - `uvicorn --workers 4` and Prometheus scrapes whichever of the four children the socket - happens to hand it, so counters appear to jump backwards. There are two fixes: - `PROMETHEUS_MULTIPROC_DIR` + `MultiProcessCollector` (a shared mmap directory, a - gauge-mode decision at every call site, and dead files to garbage-collect after every - crash), or one process per pod and scale with replicas. This repo takes the second. - `PROMETHEUS_MULTIPROC_DIR` is deliberately not set, and nothing here reads it. +3. **One process per pod.** prometheus_client keeps its registry in process memory, so + `uvicorn --workers 4` has Prometheus scraping whichever child the socket hands it and + counters appear to jump backwards. The alternative fix — `PROMETHEUS_MULTIPROC_DIR` and + `MultiProcessCollector` — costs a shared mmap directory, a gauge-mode decision at every + call site, and dead files to collect after every crash. This repo scales with replicas + instead; nothing here reads that variable. """ from __future__ import annotations @@ -52,9 +46,8 @@ if TYPE_CHECKING: # --- Metrics ---------------------------------------------------------------------------- # # Module level, created exactly once at import. A second registration of the same name -# against the default registry raises ValueError, which is a feature: it turns "two modules -# each defined their own copy of this counter" into an ImportError at startup instead of a -# metric that silently reports half the truth. +# raises ValueError, which turns "two modules each defined their own copy" into a startup +# failure instead of a metric that silently reports half the truth. TASKS_CLAIMED = Counter( "svcforge_tasks_claimed_total", @@ -79,9 +72,8 @@ TASKS_DEAD_LETTERED = Counter( PROVISION_TIME = Histogram( "svcforge_provision_duration_seconds", "Wall time of a provision task, claim to terminal report.", - # NOT the defaults. See the module docstring: the defaults end at 10s and a provision - # takes minutes. The top finite bucket is 1800 because helm's own --timeout is 600 and - # a provision past thirty minutes is broken and belongs in +Inf. + # Not the defaults — see the module docstring. The top finite bucket is 1800 because + # helm's own --timeout is 600, so a provision past thirty minutes belongs in +Inf. buckets=(10, 30, 60, 120, 300, 600, 1800, float("inf")), ) @@ -123,9 +115,9 @@ def _add_trace_ids( ) -> structlog.typing.EventDict: """Stamp the active trace/span id onto the line, if there is one. - This is the join key. Without it, "find the logs for this trace" is a full-text search - over a time window and a guess; with it, it is one query. Hex-formatted to the widths - the W3C spec uses, so the value pasted from Tempo matches the value in Loki. + The join key: without it, "find the logs for this trace" is a full-text search over a + time window and a guess. Hex-formatted to the W3C widths, so a value pasted from Tempo + matches the value in Loki. """ span = trace.get_current_span() ctx = span.get_span_context() @@ -138,10 +130,9 @@ def _add_trace_ids( def setup(service_name: str, settings: Settings) -> None: """Configure structlog, the tracer provider, and the metric registry. Idempotent. - Called once from each service's entrypoint, before anything else — "before anything - else" because any logger bound before this runs keeps the default configuration - (`cache_logger_on_first_use`), and a module-level `log = structlog.get_logger()` in an - import that lands first will print unstructured text forever. + Called once from each service's entrypoint, before anything else: a logger bound before + this runs keeps the default configuration (`cache_logger_on_first_use`), so a + module-level `log = structlog.get_logger()` prints unstructured text forever. """ global _configured # process-wide config is process-wide state if _configured: @@ -155,9 +146,9 @@ def setup(service_name: str, settings: Settings) -> None: def _setup_logging(service_name: str, settings: Settings) -> None: """structlog + stdlib logging, both rendering JSON to stdout through one handler. - The stdlib half is not optional. `psycopg`, `httpx`, `uvicorn` and the OTEL SDK all log - through `logging`; without the ProcessorFormatter bridge below, their lines arrive as - bare text on the same stdout and every one of them is a parse failure in the collector. + The stdlib half is not optional: `psycopg`, `httpx`, `uvicorn` and the OTEL SDK all log + through `logging`, and without the ProcessorFormatter bridge their lines arrive as bare + text on the same stdout — a parse failure each in the collector. """ level = getattr(logging, settings.log_level.upper(), logging.INFO) @@ -200,14 +191,13 @@ def _setup_logging(service_name: str, settings: Settings) -> None: root = logging.getLogger() # Replace rather than append: basicConfig may already have run, and two handlers means - # two copies of every line. stdout only — a container writes logs to stdout and the - # collector tails them from there. A log file inside a pod is deleted with the pod. + # two copies of every line. stdout only — a log file inside a pod dies with the pod. root.handlers = [handler] root.setLevel(level) - # Remembered so bind_task_context can restore it after clearing. Without this, every - # log line emitted inside a task loses `service`, and those are exactly the lines you - # filter on when you are trying to tell worker output from reconciler output. + # Remembered so bind_task_context can restore it after clearing. Without it every line + # emitted inside a task loses `service`, which is what tells worker output from + # reconciler output. global _service_name _service_name = service_name structlog.contextvars.bind_contextvars(service=service_name) @@ -216,10 +206,9 @@ def _setup_logging(service_name: str, settings: Settings) -> None: def _setup_tracing(service_name: str, settings: Settings) -> None: """Set the global tracer provider, exporting over OTLP when an endpoint is configured. - Skipped entirely when something already set a provider: the API runs under - `opentelemetry-instrument`, whose auto-instrumentation installs one before our - `main()` is reached. Overwriting it drops the FastAPI and psycopg instrumentation's - spans on the floor, and the SDK only logs a warning about it. + Skipped when something already set one: the API runs under `opentelemetry-instrument`, + which installs a provider before `main()` is reached. Overwriting it drops the FastAPI + and psycopg spans on the floor, and the SDK only logs a warning. """ if isinstance(trace.get_tracer_provider(), TracerProvider): return @@ -230,7 +219,7 @@ def _setup_tracing(service_name: str, settings: Settings) -> None: exporter = _otlp_exporter(settings.otel_endpoint) if exporter is not None: # Batch, not Simple: SimpleSpanProcessor exports inline on span end, so every - # helm span would block on a network round trip to the collector. + # helm span would block on a round trip to the collector. provider.add_span_processor(BatchSpanProcessor(exporter)) trace.set_tracer_provider(provider) @@ -239,9 +228,9 @@ def _setup_tracing(service_name: str, settings: Settings) -> None: def _otlp_exporter(endpoint: str) -> Any | None: # noqa: ANN401 - one of two exporter classes """The OTLP exporter, if the optional exporter package is installed. - Optional on purpose. In the cluster the API runs under `opentelemetry-instrument`, - which brings its own exporter and configures it from `OTEL_EXPORTER_OTLP_*`. Making it - a hard dependency of the shared library would mean every unit test imports gRPC. + Optional on purpose: in the cluster the API runs under `opentelemetry-instrument`, which + brings its own exporter configured from `OTEL_EXPORTER_OTLP_*`. As a hard dependency of + the shared library it would make every unit test import gRPC. """ try: from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter @@ -263,8 +252,8 @@ def get_logger(name: str) -> structlog.stdlib.BoundLogger: def tracer() -> trace.Tracer: """The svcforge tracer. Manual spans wrap helm calls, and nothing else. - Everything else is auto-instrumented (FastAPI, psycopg). A hand-rolled span around a - function that the SDK already wraps is a duplicated span and a maintenance cost. + FastAPI and psycopg are auto-instrumented, and a hand-rolled span around something the + SDK already wraps is a duplicate to maintain. """ return trace.get_tracer(_TRACER_NAME) @@ -285,17 +274,15 @@ def start_metrics_server(port: int) -> None: def bind_task_context(instance_id: UUID, task_id: int, team: str) -> None: """Bind the three keys every log line in a task must carry. Called at claim time. - `clear_contextvars()` first, and this is the whole reason the function exists rather - than three `bind_contextvars` calls at the call site. A worker coroutine reuses its - context across loop iterations; without the clear, task 41's `instance_id` is still - bound when task 42 starts logging, and the log for the incident you are debugging - names the wrong tenant. Contextvars are per-task in asyncio, which makes this safe - under the concurrency semaphore: two handlers running at once do not see each other's. + `clear_contextvars()` first, which is why this is a function rather than three + `bind_contextvars` calls at the call site: a worker coroutine reuses its context across + iterations, so without the clear, task 41's `instance_id` is still bound when task 42 + logs and the incident names the wrong tenant. Contextvars are per-task in asyncio, so + two handlers under the concurrency semaphore do not see each other's. """ structlog.contextvars.clear_contextvars() structlog.contextvars.bind_contextvars( - # `service` is re-bound because the clear above took it with it. It is set once in - # setup() and is not per-task, but clear_contextvars() is indiscriminate. + # Re-bound because the indiscriminate clear above took it; it is not per-task. service=_service_name, instance_id=str(instance_id), task_id=task_id, @@ -307,8 +294,7 @@ def inject_traceparent() -> str | None: """Serialise the active span context to a W3C traceparent, for the tasks row. None when there is no recording span — a task enqueued by the reconciler's own tick has - no inbound request to be part of. Nullable column, nullable return: an untraced task is - normal, not an error. + no inbound request to belong to. Nullable column, nullable return. """ carrier: dict[str, str] = {} _propagator.inject(carrier) @@ -318,9 +304,9 @@ def inject_traceparent() -> str | None: def context_from_traceparent(traceparent: str | None) -> Context: """Inverse of inject_traceparent. Used at claim to parent the worker span to the API's. - An empty Context for None or for a malformed value — `extract` does not raise on a - traceparent that fails to parse, it returns the carrier's context unchanged, and the - resulting span starts a new trace. A bad header must never fail a provision. + An empty Context for None or for a malformed value: `extract` does not raise on an + unparseable traceparent, it returns the carrier's context unchanged and the span starts + a new trace. A bad header must never fail a provision. """ if not traceparent: return Context() diff --git a/libs/svcforge_core/svcforge_core/repo/db.py b/libs/svcforge_core/svcforge_core/repo/db.py index 8215337..7b17cf0 100644 --- a/libs/svcforge_core/svcforge_core/repo/db.py +++ b/libs/svcforge_core/svcforge_core/repo/db.py @@ -12,16 +12,15 @@ from psycopg import AsyncConnection from psycopg.rows import dict_row from psycopg_pool import AsyncConnectionPool -# The pool hands out dict-row connections because of `row_factory=dict_row` below. Say so -# in the type system too, or every `row["attempts"]` in this codebase is a mypy error -# 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. +# The cap on error text written to `instances.error` and `tasks.last_error`. A helm failure +# can emit megabytes and these columns are read by humans. Defined once so the two call +# paths that feed the same columns agree. ERROR_MAX_CHARS = 2000 +# The pool hands out dict-row connections because of `row_factory=dict_row` below, and the +# type system has to say so: against a bare `AsyncConnectionPool`, which resolves to tuple +# rows, every `row["attempts"]` is a mypy error. The reach-for fix is `# type: ignore`, +# which throws away the checking entirely. type DictRow = dict[str, Any] type DictConnection = AsyncConnection[DictRow] type DictPool = AsyncConnectionPool[DictConnection] @@ -30,28 +29,25 @@ type DictPool = AsyncConnectionPool[DictConnection] def make_pool(dsn: str, min_size: int = 1, max_size: int = 5) -> DictPool: """Construct the pool. Does NOT open it — the caller owns open/close. - `open=False` is deliberate: the constructor does zero I/O, so building a pool at - import time and never opening it fails later as a PoolTimeout at first use, far - from the cause. The caller (a FastAPI lifespan, a worker main) opens and closes it. + `open=False` because the constructor does zero I/O: a pool built at import time and + never opened fails later as a PoolTimeout at first use, far from the cause. The caller + (a FastAPI lifespan, a worker main) opens and closes it. - kwargs are per-connection: + Two per-connection kwargs matter: - * `prepare_threshold=None` — REQUIRED through pgbouncer in transaction mode. - psycopg3 auto-prepares a statement after it sees it 5 times. pgbouncer may hand - the next execution to a different backend, which has never heard of that prepared - statement. Symptom: everything works for exactly five calls, then - `prepared statement "_pg3_0" does not exist` — intermittent, only under - concurrency, never in a unit test. - - * `row_factory=dict_row` — rows arrive as dicts, so `Instance.model_validate(row)` - works directly instead of unpacking tuples by position. + * `prepare_threshold=None` — REQUIRED through pgbouncer in transaction mode. psycopg3 + auto-prepares a statement after five executions, and pgbouncer may hand the sixth to a + backend that has never heard of it. Symptom: `prepared statement "_pg3_0" does not + exist`, intermittent, only under concurrency, never in a unit test. + * `row_factory=dict_row` — rows arrive as dicts, so `Instance.model_validate(row)` works + without unpacking tuples by position. Also gone on 6543: LISTEN/NOTIFY, session-level SET, cross-statement advisory locks. - `SELECT ... FOR UPDATE SKIP LOCKED` inside one transaction is unaffected — which is - exactly why the queue is built on it. Use the session pooler (5432) for migrations. + `SELECT ... FOR UPDATE SKIP LOCKED` inside one transaction is unaffected, which is why + the queue is built on it. Migrations use the session pooler (5432). max_size is a database-capacity decision, not a throughput knob: the free tier has a - small connection budget, and replicas multiply this number. + small connection budget and replicas multiply this number. """ return AsyncConnectionPool( conninfo=dsn, diff --git a/libs/svcforge_core/svcforge_core/repo/reconcile.py b/libs/svcforge_core/svcforge_core/repo/reconcile.py index 04ed9da..86525ee 100644 --- a/libs/svcforge_core/svcforge_core/repo/reconcile.py +++ b/libs/svcforge_core/svcforge_core/repo/reconcile.py @@ -1,17 +1,14 @@ """The reconciler's SQL. -Why this file exists rather than the queries living in `services/reconciler/main.py`: the -layer rule says transport knows nothing about SQL, and the reconciler is transport — a CLI -entrypoint. It gets its own repo module rather than growing `InstanceRepo` and `TaskRepo` -because everything here is a *sweep*: it reads rows nobody asked about and it writes an -instance state and a task row in the same transaction. `InstanceRepo.update_state` owns its -own connection by design, so the reconciler cannot get atomicity from it without reaching -around the repo — which is the thing the layer rule exists to prevent. +Its own module rather than queries in `services/reconciler/main.py` (transport knows no +SQL) and rather than more methods on `InstanceRepo`/`TaskRepo`, because everything here is +a *sweep*: it reads rows nobody asked about and writes an instance state and a task row in +one transaction. `InstanceRepo.update_state` owns its own connection by design, so the +reconciler cannot get that atomicity without reaching around the repo. -The recurring shape below is: lock the row, re-check the condition under the lock, act. -The re-check is not paranoia about concurrency — the reconciler is a singleton. It is what -makes the sweep idempotent against *itself*: a tick that crashes after the insert and -before the commit must leave nothing behind, and the next tick must not double-enqueue. +The recurring shape is: lock the row, re-check the condition under the lock, act. The +re-check makes the sweep idempotent against *itself* — the reconciler is a singleton, but a +tick that crashes before its commit must leave nothing behind for the next one to double. """ from __future__ import annotations @@ -28,10 +25,9 @@ from svcforge_core.obs import inject_traceparent 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 -# deprovision that exhausted its attempts must be re-enqueueable by the next sweep, or a -# transient cluster outage would permanently strand the instance. +# What "already outstanding" means to the idempotency guard on every enqueue below. +# 'done'/'failed' are not outstanding: a deprovision that exhausted its attempts must be +# re-enqueueable, or a transient cluster outage strands the instance permanently. _UNFINISHED = (TaskState.QUEUED.value, TaskState.RUNNING.value) @@ -46,9 +42,8 @@ class ReconcileRepo: async def queue_depth(self) -> int: """Tasks waiting to be claimed. - Counts every `queued` row, not just the runnable ones (`run_after <= now()`). The - alert on this gauge is `deriv(...) > 0` — "the backlog is growing" — and a backlog - of tasks parked on backoff is exactly the backlog you want to see growing. + Every `queued` row, not just the runnable ones. The alert is `deriv(...) > 0` — "the + backlog is growing" — and tasks parked on backoff are part of that backlog. """ async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute("select count(*) as n from tasks where state = %s", (TaskState.QUEUED.value,)) @@ -77,9 +72,9 @@ class ReconcileRepo: async def known_releases(self) -> set[tuple[str, str]]: """(release_name, namespace) for every instance row, in any state. - Any state, deliberately. An instance that is still `requested` has no release yet, - but a worker may be installing it *right now* — treating it as unknown would - report a healthy in-flight provision as an orphan on every tick. + Any state, deliberately. A `requested` instance has no release yet, but a worker may + be installing it right now, and treating it as unknown reports a healthy in-flight + provision as an orphan. """ async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute("select release_name, namespace from instances") @@ -91,20 +86,16 @@ class ReconcileRepo: None if the row moved, or if a provision is already outstanding. - The two-hop state change is the interesting part: + The state change takes two hops. `LEGAL` has no `ready -> provisioning` edge — the + tenant-visible lifecycle leaves `ready` only through `deleting` or `failed`, and + drift is a failure — so it goes `ready -> failed -> provisioning`, both edges legal + and asserted below rather than assumed. It has to land in `provisioning`, not + `failed`: `handle_provision` finishes with a `provisioning -> ready` CAS, and given a + `failed` row helm runs, the CAS matches nothing, and the instance sits in `failed` + forever with a healthy release behind it. - * `LEGAL` has no `ready -> provisioning` edge. The tenant-visible lifecycle only - leaves `ready` through `deleting` or `failed`, and drift is a failure — the - service the tenant is paying for is gone. So: `ready -> failed -> provisioning`, - both edges legal, asserted below by the domain function rather than assumed. - * The row must land in `provisioning`, not `failed`, before the worker sees the - task. `handle_provision` CASes `requested -> provisioning` best-effort and then - CASes `provisioning -> ready` for real; hand it a `failed` row and helm runs, the - final CAS matches nothing, and the instance sits in `failed` forever with a - healthy release behind it. - - Both hops and the insert are one transaction, so the row is never observable in the - intermediate `failed` state and a crash mid-sweep leaves nothing half-done. + Both hops and the insert are one transaction, so the intermediate `failed` is never + observable and a crash mid-sweep leaves nothing half-done. """ async with self._pool.connection() as conn: async with conn.transaction(), conn.cursor() as cur: @@ -117,8 +108,8 @@ class ReconcileRepo: if await _has_unfinished(cur, instance_id, TaskKind.PROVISION): return None - # Assert the path through the state machine instead of trusting the SQL. - # If someone edits LEGAL, this raises here rather than corrupting rows. + # Assert the path through the state machine instead of trusting the SQL: an + # edit to LEGAL raises here rather than corrupting rows. failed = transition(InstanceState.READY, InstanceState.FAILED) provisioning = transition(failed, InstanceState.PROVISIONING) @@ -135,17 +126,15 @@ class ReconcileRepo: Two populations, one query: - * `ready` and past `expires_at` — the TTL sweep proper. The whole reason a - throwaway Elasticsearch does not become a permanent line on the cloud bill. - * `deleting` with nothing to do the deleting — the API CASes to `deleting` and then + * `ready` past `expires_at` — the TTL sweep, which is what stops a throwaway + Elasticsearch becoming a permanent line on the cloud bill. + * `deleting` with nothing doing the deleting — the API CASes to `deleting` and enqueues in a second statement, and a crash between the two leaves exactly this. - That ordering is chosen *because* this sweep exists; the other order would leave - a deprovision task pointing at a `ready` instance, and a worker would tear down a - live service nobody asked to delete. + That order is chosen because this sweep exists; the reverse would leave a + deprovision task on a `ready` instance and tear down a live service. - Note the parentheses around the OR. Without them, `and not exists (...)` binds to - the second branch alone and the query re-enqueues a deprovision for every deleting - instance on every tick, forever. + Note the parentheses around the OR: without them `and not exists (...)` binds to the + second branch alone and every deleting instance is re-enqueued on every tick. """ async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( @@ -170,10 +159,10 @@ class ReconcileRepo: async def enqueue_deprovision(self, instance_id: UUID) -> int | None: """CAS to `deleting` if needed, and enqueue the task. One transaction. None if moot. - The instance must be in `deleting` before the worker claims the task, for the same - reason as `enqueue_reprovision`: `handle_deprovision` finishes with a - `deleting -> deleted` CAS, and a `ready` row would make helm uninstall the release - and the DB keep advertising an endpoint that no longer resolves. + The instance must reach `deleting` before the worker claims the task, for the same + reason as `enqueue_reprovision`: `handle_deprovision` ends with a `deleting -> + deleted` CAS, and on a `ready` row helm uninstalls the release while the DB keeps + advertising an endpoint that no longer resolves. """ async with self._pool.connection() as conn: async with conn.transaction(), conn.cursor() as cur: @@ -208,15 +197,13 @@ class ReconcileRepo: """Enqueue an upgrade unless one is already outstanding. None if it is. The guard is what keeps the fleet at `max_in_flight`. The work list is a query over - `chart_version`, and that column is only written *after* helm reports success — so - an instance stays on the work list for the entire duration of its own upgrade, and - for the hours it spends parked waiting for its 03:00 window. Without this check the - sweep enqueues one more upgrade for the same instance every 60 seconds, and - `max_in_flight=1` becomes sixty tasks an hour against one release. + `chart_version`, which is written only after helm reports success, so an instance + stays on the list for the whole duration of its own upgrade and for the hours it + spends parked waiting for its 03:00 window. Without the check, `max_in_flight=1` + becomes sixty tasks an hour against one release. - `verify` counts as outstanding too: an upgrade whose verify has not reported is an - upgrade still in progress, and re-enqueueing it would race the probe that decides - whether the whole rollout halts. + `verify` counts as outstanding too: re-enqueueing an upgrade whose verify has not + reported would race the probe that decides whether the rollout halts. """ async with self._pool.connection() as conn: async with conn.transaction(), conn.cursor() as cur: @@ -232,9 +219,8 @@ async def _has_unfinished( ) -> bool: """Is a task of any of these kinds queued or running for this instance? - Takes the caller's cursor on purpose: the answer is only true for as long as the - transaction that asked, and checking on a separate connection would be a check against - a different snapshot than the insert that follows it. + Takes the caller's cursor: the answer holds only for the asking transaction, and a + separate connection would check a different snapshot than the insert that follows. """ await cur.execute( """select 1 from tasks @@ -258,10 +244,9 @@ async def _insert_task( """Insert one task in the caller's transaction, carrying the current trace context. `traceparent` is written here rather than left to `TaskRepo.enqueue` because these rows - are inserted inside a transaction the reconciler owns. Nothing propagates a trace - through a table on its own — see `obs.inject_traceparent`. It is null when the sweep is - not itself inside a span, which is fine and expected: a nullable column for an untraced - task. + go in inside a transaction the reconciler owns, and nothing propagates a trace through a + table on its own (see `obs.inject_traceparent`). Null when the sweep is not itself inside + a span, which is normal. """ await cur.execute( """insert into tasks (instance_id, kind, run_after, traceparent) diff --git a/libs/svcforge_core/svcforge_core/repo/tasks.py b/libs/svcforge_core/svcforge_core/repo/tasks.py index 64ee488..4644196 100644 --- a/libs/svcforge_core/svcforge_core/repo/tasks.py +++ b/libs/svcforge_core/svcforge_core/repo/tasks.py @@ -1,11 +1,8 @@ -"""The queue. +"""The queue: a Postgres table, not Redis. -The queue is a Postgres table, not Redis. The reason is one sentence: a task and the -instance state it describes must commit atomically. Split them across two stores and you -own a distributed commit problem that has no winning move — the process can die between -the two writes, and whichever you write first is the one that lies. - -Everything else here follows from that. +A task and the instance state it describes must commit atomically. Split across two stores +that is a distributed commit problem with no winning move — the process can die between the +two writes, and whichever went first is the one that lies. Everything here follows from that. """ from __future__ import annotations @@ -22,40 +19,30 @@ from svcforge_core.domain.states import LEGAL, InstanceState from svcforge_core.obs import TASKS_DEAD_LETTERED, inject_traceparent 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` -# instance to `failed` — a transition domain.transition() explicitly forbids, performed -# by raw SQL that never asks it. The state machine has to be the same one everywhere, or -# it is decoration. +# Which states may legally become `failed`, derived from the domain's table rather than +# restated here. Without this guard the UPDATE below would move a `deleted` instance to +# `failed` — a transition domain.transition() forbids, performed by SQL that never asks it. _CAN_FAIL: Final[tuple[str, ...]] = tuple( state.value for state, allowed in LEGAL.items() if InstanceState.FAILED in allowed ) # The claim query. Do not "simplify" this into two statements. # -# Postgres has no `UPDATE ... LIMIT`, so the row is chosen by a subquery. That subquery -# takes a row lock (`for update`) and steps over rows other workers already hold -# (`skip locked`) instead of blocking behind them — which is what makes N workers scale -# instead of queueing single-file behind the oldest task. +# Postgres has no `UPDATE ... LIMIT`, so a subquery picks the row. It takes a row lock +# (`for update`) and steps over rows other workers hold (`skip locked`) instead of blocking +# behind them, which is what lets N workers scale instead of queueing behind the oldest +# task. Select-then-update as two statements leaves a gap where a second worker reads the +# same id and both provision — small enough to miss in testing and hit in production. # -# The whole thing is ONE statement on purpose. Select-then-update as two statements -# leaves a gap in which a second worker reads the same id, and both provision. The gap -# is small, which means you will not hit it in testing and will hit it in production. +# The `with claimed as (...)` wrapper changes nothing about the locking: the UPDATE and its +# subquery are still one statement, and a data-modifying CTE runs exactly once. The outer +# SELECT only joins `instances.team` onto the claimed row so the worker can bind `team` to +# its log context without a second round trip. # -# The `with claimed as (...)` wrapper is the ONLY addition to the canonical form, and it -# changes nothing about the locking: the UPDATE and its `for update skip locked` subquery -# are still one statement, executed once. The outer SELECT only joins `instances.team` -# onto the row that was already claimed, so the worker can bind `team` to its log context -# before it has loaded anything. A data-modifying CTE runs exactly once and cannot claim -# twice. The alternative — a second SELECT for the team — would be a second round trip per -# task to fetch a column the database already had in hand. -# -# LEFT join, not inner. The UPDATE inside the CTE has already taken effect by the time the -# outer select runs, so an inner join that matches nothing would return no row — and -# `claim()` would report "queue empty" for a task it had just marked `running`, stranding -# it until the lease expires and silently burning an attempt. The FK cascade makes that -# nearly impossible in practice; "nearly" is not a reason to leave a silent failure in the -# one query the whole system depends on. `Task.team` is already `str | None`. +# LEFT join, not inner. The CTE's UPDATE has already taken effect when the outer select +# runs, so an inner join matching nothing returns no row — `claim()` would report "queue +# empty" for a task it just marked `running`, stranding it until the lease expires and +# burning an attempt. `Task.team` is already `str | None`. _CLAIM_SQL = """ with claimed as ( update tasks set state='running', attempts=attempts+1, locked_by=%(worker)s, locked_at=now() @@ -88,16 +75,14 @@ class TaskRepo: ) -> Task: """Insert a task inside the CALLER's transaction. - Takes `conn` so the API can insert the instance and enqueue its provision task in - one transaction. Rolling back must lose both, or you get an orphan task pointing - at an instance that was never committed. + Takes `conn` so the API can insert the instance and enqueue its provision task + together. A rollback must lose both, or an orphan task points at an instance that + was never committed. - The `traceparent` is captured here, at enqueue time, because this is the last - moment the caller's span context still exists. Trace context does NOT survive a - queue on its own: the worker picks the row up in a different process, minutes - later, with no ambient context. Writing the W3C traceparent onto the row is the - thread that lets the worker re-parent its span to the POST that caused it — the - difference between one trace spanning API → queue → helm and two unrelated ones. + `traceparent` is captured here because this is the last moment the caller's span + context exists. Trace context does not survive a queue on its own — the worker picks + the row up in another process minutes later — so writing the W3C traceparent onto + the row is what lets it re-parent its span to the POST that caused it. """ async with conn.cursor() as cur: await cur.execute( @@ -118,10 +103,8 @@ class TaskRepo: ) -> int: """Enqueue in its own transaction, returning the new task id. - For callers with nothing to commit alongside it — the reconciler, tests. The - module specs disagree about enqueue's shape (Module 2 passes a conn, Module 4 - does not); rather than making `conn` optional and quietly hiding the transaction - question, both callers get an honest method name. + For callers with nothing to commit alongside it — the reconciler, tests. A separate + method rather than an optional `conn`, which would hide the transaction question. """ async with self._pool.connection() as conn: task = await self.enqueue(conn, instance_id, kind, run_after) @@ -151,12 +134,11 @@ class TaskRepo: and recording what it accomplished belong in one transaction, or a crash between them leaves a task marked done whose work never landed. - `and state='running' and locked_by=%s` is not defensive padding — without it this - is a lost-update bug with a real trigger. A worker that hangs past `lease_seconds` - has its task requeued by the reconciler and re-claimed by someone else. When the - hung worker finally returns, an unconditional UPDATE here marks the task `done` - while the new owner is still running it, and its work goes unaccounted for. The - loser gets False and must treat it as "someone else owns this now", not an error. + `and state='running' and locked_by=%s` is a lost-update guard with a real trigger. A + worker that hangs past `lease_seconds` has its task requeued and re-claimed; when it + returns, an unconditional UPDATE marks the task `done` while the new owner is still + running it. The loser gets False and treats it as "someone else owns this", not an + error. """ sql = "update tasks set state='done', locked_by=null where id=%s and state='running' and locked_by=%s" if conn is not None: @@ -170,17 +152,16 @@ class TaskRepo: async def fail(self, task_id: int, err: str, worker_id: str, max_attempts: int = 5) -> bool: """Retry with backoff, or give up. False if this worker no longer owns the task. - Under max_attempts: back to 'queued' with run_after pushed out by exponential - backoff with full jitter. Jitter matters — a cluster-wide outage fails every task - at once, and without it every worker retries in the same instant, forever. + Under max_attempts: back to 'queued', run_after pushed out by exponential backoff + with full jitter. Jitter matters — a cluster-wide outage fails every task at once, + and without it every worker retries in the same instant, forever. - At max_attempts: 'failed', and the error is copied onto the instance so the tenant - can see it. A dead-letter state, not an infinite retry: a task that cannot succeed - must stop and become someone's problem. + At max_attempts: 'failed', a dead-letter state rather than an infinite retry, and + for a provision the error is copied onto the instance so the tenant can see it. - The ownership check in the SELECT is the same lost-lease guard as `complete`, and - it matters more here: a stale worker reporting failure would push a task the new - owner is actively running back to `queued`, letting a *third* worker claim it. + The ownership check is the same lost-lease guard as `complete`, and matters more + here: a stale worker reporting failure would push a task the new owner is running + back to `queued`, letting a third worker claim it. """ now = datetime.now(UTC) async with self._pool.connection() as conn: @@ -193,8 +174,8 @@ class TaskRepo: ) row = await cur.fetchone() if row is None: - # Either the task is gone, or the lease was stolen. Both mean: not ours - # to report on. Writing anything here would corrupt the new owner's run. + # Task gone, or the lease was stolen. Either way it is not ours to + # report on, and writing here would corrupt the new owner's run. return False attempts = int(row["attempts"]) instance_id = row["instance_id"] @@ -215,22 +196,20 @@ class TaskRepo: where id = %s""", (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 - # 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. + # Dead-lettering the task is right for every kind; moving the INSTANCE to + # `failed` is right only for provision, where nothing but a human recovers + # it. For the other three the instance is still healthy and something else + # owns recovery: + # deprovision — stays `deleting`, which is what lets due_for_deprovision + # re-enqueue it. `failed` drops it out of that query and + # leaks the release forever. + # upgrade — helm --atomic rolled back, so it is `ready` on the previous + # version. check_version_drift retries next window; `failed` + # would drop it off the upgrade work-list. + # verify — handle_verify already halted the rollout; drift + # re-provisions if the release vanished. + # The dead-letter metric and its alert cover 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() @@ -245,10 +224,9 @@ class TaskRepo: async def reset_expired_leases(self, lease_seconds: int) -> int: """Return tasks whose worker died back to the queue. Called by the reconciler. - No distributed lock survives a power cut. A worker that is SIGKILLed leaves - `state='running'` and `locked_by` set with nobody running it, and that row would - sit there forever. The lease is the only thing that recovers it, which is why - `locked_at` exists. + No distributed lock survives a power cut. A SIGKILLed worker leaves `state='running'` + and `locked_by` set with nobody running it, and that row sits there forever. The + lease is the only thing that recovers it, which is why `locked_at` exists. """ async with self._pool.connection() as conn, conn.cursor() as cur: await cur.execute( diff --git a/libs/svcforge_core/svcforge_core/runtime.py b/libs/svcforge_core/svcforge_core/runtime.py index dc703b0..cb89f6d 100644 --- a/libs/svcforge_core/svcforge_core/runtime.py +++ b/libs/svcforge_core/svcforge_core/runtime.py @@ -26,11 +26,9 @@ async def sleep_or_stop(stop: asyncio.Event, seconds: float) -> None: 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. + add_signal_handler, not signal.signal: the latter runs 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. """ loop = asyncio.get_running_loop() for sig in (signal.SIGTERM, signal.SIGINT): diff --git a/libs/svcforge_core/svcforge_core/settings.py b/libs/svcforge_core/svcforge_core/settings.py index a55faa4..ed3fbb1 100644 --- a/libs/svcforge_core/svcforge_core/settings.py +++ b/libs/svcforge_core/svcforge_core/settings.py @@ -67,12 +67,8 @@ class Settings(BaseSettings): kubectl_bin: str = "kubectl" helm_timeout_s: float = Field(default=300.0, gt=0) - # --- CLI ---------------------------------------------------------------------- - # The CLI is an API client and nothing more. It gets a URL and a token; it does not - # get a DSN, because the moment a human can reach the database directly, someone will - # "just fix one row" and the state machine stops being true. - api_url: str = "http://localhost:8000" - api_token: str | None = None + # The CLI's own settings live in `services/cli/main.py`, not here: this model requires + # SVCFORGE_PG_DSN, and the CLI is an API client that must never hold one. # --- Observability ------------------------------------------------------------ log_level: str = "info" @@ -87,8 +83,8 @@ class Settings(BaseSettings): 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. + A property so the three entrypoints do not each pick between `str(pg_dsn)` and + `pg_dsn.unicode_string()`, which had drifted across the services. """ return str(self.pg_dsn) @@ -100,13 +96,11 @@ class Settings(BaseSettings): def check_production(self) -> None: """Refuse the dev escape hatches outside local development. Call at startup. - This is a no-op unless `SVCFORGE_ENVIRONMENT` says otherwise, which is what makes - it safe to call unconditionally from every entrypoint — and calling it - unconditionally is the point. The previous version could only be invoked from a - branch that already knew it was production, so no such branch was ever written and - the check never ran: `SVCFORGE_AUTH_DISABLED=true` in prod would have started the - API with JWT verification off, serving every unauthenticated request as team - `platform`, silently. + A no-op unless `SVCFORGE_ENVIRONMENT` says otherwise, which is what makes it safe to + call unconditionally — and unconditionally is the point. A version invoked only from + a branch that already knew it was production never ran at all, and + `SVCFORGE_AUTH_DISABLED=true` in prod would silently serve every unauthenticated + request as team `platform`. """ if self.environment == "local": return diff --git a/services/api/deps.py b/services/api/deps.py index e156ca9..a17f3e3 100644 --- a/services/api/deps.py +++ b/services/api/deps.py @@ -1,8 +1,8 @@ """Dependency injection: how a handler gets a pool, a repo, a catalog, and a team. Everything expensive — the pool, the JWKS client, the parsed catalog — is built once in -`lifespan` and parked on `app.state`. These functions only hand it out. A `Depends` that -does I/O per request is a `Depends` that does that I/O on every request forever. +`lifespan` and parked on `app.state`; these functions only hand it out. A `Depends` that +does I/O per request does that I/O on every request forever. """ from __future__ import annotations @@ -22,10 +22,9 @@ from svcforge_core.repo.instances import InstanceRepo from svcforge_core.repo.tasks import TaskRepo from svcforge_core.settings import Settings -# The algorithm allow-list is the whole point of naming algorithms explicitly. -# `jwt.decode(..., algorithms=...)` without it accepts whatever the *token* claims in its -# own header — including `none`, and including HS256 verified with the RSA public key as -# an HMAC secret. Both are forgery. The list is not configuration. +# The algorithm allow-list is not configuration. Without it, `jwt.decode` accepts whatever +# the *token* claims in its own header — including `none`, and including HS256 verified +# with the RSA public key as an HMAC secret. Both are forgery. ALLOWED_ALGORITHMS = ["RS256"] # What `auth_disabled` returns. Settings.check_production() refuses that flag in prod. @@ -42,9 +41,9 @@ _bearer = HTTPBearer(auto_error=False) def _unauthorized() -> HTTPException: """One shape for every auth failure. - Expired, wrong issuer, wrong audience, bad signature, malformed, no header: all the - same 401 with the same body. Telling a caller *which* one turns the endpoint into an - oracle they can tune a forgery against. + Expired, wrong issuer, wrong audience, bad signature, malformed, no header: the same 401 + with the same body. Naming which one turns the endpoint into an oracle a forger can tune + against. """ return HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, @@ -68,8 +67,8 @@ async def get_pool(request: Request) -> DictPool: def get_catalog(request: Request) -> dict[str, CatalogEntry]: """The catalog, parsed once at startup. - Read from disk per request and a mid-flight edit to catalog.yaml changes the answer - between two requests of the same deploy. Load it at startup; a change is a restart. + Read per request, a mid-flight edit to catalog.yaml would change the answer between two + requests of the same deploy. A catalog change is a restart. """ catalog: dict[str, CatalogEntry] = request.app.state.catalog return catalog @@ -102,16 +101,14 @@ async def get_current_team( jwks_client: PyJWKClient | None = getattr(request.app.state, "jwks_client", None) if jwks_client is None: - # Auth is on but there is no key source. Fail closed. Answering 500 here would be - # honest about the cause and would also let a misconfigured deploy be told apart - # from a bad token; 401 is the same answer a forger gets. + # Auth is on but there is no key source. Fail closed. A 500 would be honest about + # the cause and would also let a forger tell a misconfigured deploy from a bad token. raise _unauthorized() try: - # PyJWKClient keeps its own TTL cache, so this is a dict lookup on the hot path. - # It is only blocking on a cache MISS (key rotation) — hence to_thread, which - # costs a thread hop we take a handful of times a day rather than an event loop - # stalled on someone else's HTTP call once per rotation. + # PyJWKClient keeps a TTL cache, so this is a dict lookup on the hot path and only + # blocks on a miss (key rotation) — hence to_thread, a thread hop a few times a day + # rather than an event loop stalled on someone else's HTTP call. signing_key = await _signing_key(jwks_client, creds.credentials) claims: dict[str, Any] = jwt.decode( creds.credentials, @@ -139,10 +136,9 @@ async def get_current_team( async def _signing_key(client: PyJWKClient, token: str) -> jwt.PyJWK: """Fetch the signing key off the event loop. - PyJWKClient.get_signing_key_from_jwt() does a synchronous urlopen on a cache miss. - Called directly from `async def`, that blocks the loop — every other in-flight request - on this worker stops until the identity provider answers, and if it hangs, so does the - pod, and /readyz keeps saying it is fine. + `get_signing_key_from_jwt()` does a synchronous urlopen on a cache miss. Called directly + from `async def` it blocks the loop: every other in-flight request stops until the IdP + answers, and if the IdP hangs so does the pod, with /readyz still saying it is fine. """ return await asyncio.to_thread(client.get_signing_key_from_jwt, token) @@ -159,11 +155,10 @@ async def rate_limit( ) -> None: """Per-team rate limiting. One Redis command per check, and it fails OPEN. - Failing open is the entire policy. Redis holds derived state; losing it must degrade - the platform, never stop it. A limiter that fails closed converts a cache outage into - a total outage, which is a strictly worse incident than the burst it was protecting - against — so `RateLimiter.check` swallows its own errors and returns `allowed=True`. - The 429 below therefore only ever comes from a real, counted overage. + Redis holds derived state, so losing it must degrade the platform rather than stop it: a + limiter that fails closed turns a cache outage into a total outage, a worse incident + than the burst it was guarding against. `RateLimiter.check` swallows its own errors and + returns `allowed=True`, so the 429 below only comes from a real, counted overage. """ limiter = get_rate_limiter(request) if limiter is None: diff --git a/services/api/main.py b/services/api/main.py index 2a41a91..2ba9536 100644 --- a/services/api/main.py +++ b/services/api/main.py @@ -1,7 +1,7 @@ """The app factory and its lifespan. -`create_app(settings)` is a factory, not a module-level `app = FastAPI()`, for one reason: -a test needs an app pointed at a throwaway Postgres, and an import-time app reads the real +`create_app(settings)` is a factory rather than a module-level `app = FastAPI()` because a +test needs an app pointed at a throwaway Postgres, and an import-time app reads the real environment at import time — before any fixture can say otherwise. """ @@ -32,23 +32,21 @@ log = obs.get_logger("svcforge.api") async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Open the pool, yield, close the pool. - A lifespan context, not the deprecated startup/shutdown event decorators: those cannot - express "this resource lives for exactly as long as the app", and give you no place to - put the teardown next to the setup. Closing the pool matters — an unclosed pool means - connections linger server-side after SIGTERM, and on a pooled Postgres with a small - connection budget a few rolling deploys exhaust it. + A lifespan context, not the deprecated startup/shutdown decorators: those cannot express + "this resource lives exactly as long as the app" and leave no place to put teardown next + to setup. Closing matters — an unclosed pool leaves connections open server-side after + SIGTERM, and on a pooled Postgres with a small budget a few rolling deploys exhaust it. - (The old decorator's name is spelled nowhere in this package on purpose: CI greps for - the literal string, and a comment quoting it fails the gate just as loudly as a call.) + (The old decorator's name is spelled nowhere here on purpose: CI greps for the literal + string, so a comment quoting it fails the gate as loudly as a call would.) """ settings: Settings = app.state.settings app.state.catalog = load_catalog(settings.catalog_path) - # Redis is optional by construction. `make_redis` returns None when no DSN is set, and - # every consumer treats None as "skip" — so a deployment without Redis loses rate - # limiting and keeps everything else. Built here rather than per request because a - # connection pool per request is a connection pool per request. + # Redis is optional by construction: `make_redis` returns None when no DSN is set and + # every consumer treats None as "skip", so a deployment without Redis loses rate + # limiting and keeps everything else. Built once here, not per request. redis = make_redis(settings) app.state.redis = redis app.state.rate_limiter = ( @@ -61,16 +59,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: await pool.open(wait=True) app.state.pool = pool - # The pool is open from here on, so everything below is inside the try: an exception - # in JWKS setup must still close it, or a crash-looping pod leaks a connection per - # restart until the database refuses new ones. + # The pool is open from here, so everything below is inside the try: an exception in + # JWKS setup must still close it, or a crash-looping pod leaks a connection per restart + # until the database refuses new ones. try: if settings.jwks_url and not settings.auth_disabled: client = PyJWKClient(settings.jwks_url, cache_keys=True, lifespan=300) app.state.jwks_client = client - # Warm the cache off the loop so the first authenticated request does not pay - # a blocking urlopen. Best-effort: a slow identity provider must not stop the - # pod from starting — a cache miss later just costs one to_thread hop. + # Warm the cache off the loop so the first authenticated request does not pay a + # blocking urlopen. Best-effort: a slow IdP must not stop the pod from starting, + # and a miss later costs one to_thread hop. try: await asyncio.to_thread(client.get_signing_keys) except Exception: # deliberate catch-all: startup must not hinge on the IdP being up @@ -87,9 +85,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # --------------------------------------------------------------------------- API docs -# Everything a caller needs that the generated schema cannot express on its own. Kept next -# to create_app rather than in a README because /docs is what someone integrating actually -# reads, and a README in this repo is not something they have. +# What the generated schema cannot express. Kept next to create_app because /docs is what +# someone integrating reads, and they do not have this repo. USER_GUIDE.md is the longer form. API_DESCRIPTION = """ Provision managed service instances into Kubernetes. The catalog offers Elasticsearch, Redis and Postgres, plus two deliberately tiny entries — `podinfo` and `nginx` — for @@ -144,21 +141,19 @@ OPENAPI_TAGS = [ async def _http_exception_handler(request: Request, exc: Exception) -> JSONResponse: """Render HTTPException bodies as ErrorBody, so every error has one shape. - 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. + Handlers raise `detail={"code": ..., "message": ...}`, which FastAPI's default would + nest under `{"detail": {...}}`. Plain-string details (a framework 405, say) are wrapped + so clients never 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. + Registered on starlette's HTTPException, not fastapi's. The FastAPI class 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. Keying on the parent catches + both, 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 - # through whatever a handler raised — our handlers raise dicts. Narrowing off the - # declared type would make mypy call the dict branch unreachable and delete it. + # Widened to object deliberately: Starlette types `detail` as str, but FastAPI passes + # through whatever a handler raised, and ours raise dicts. Narrowing off the declared + # type would let mypy call the dict branch unreachable and delete it. detail: object = exc.detail if isinstance(detail, dict) and "code" in detail and "message" in detail: body = ErrorBody(code=str(detail["code"]), message=str(detail["message"])) @@ -170,10 +165,9 @@ async def _http_exception_handler(request: Request, exc: Exception) -> JSONRespo 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. + A forbidden extra field, a bad type or an out-of-range ttl_days raises + RequestValidationError, which the handler above never sees. Without this, FastAPI's + default `{"detail": [...]}` is a second 422 shape alongside the handlers' ErrorBody. """ assert isinstance(exc, RequestValidationError) # noqa: S101 - registered only for this return JSONResponse( @@ -186,22 +180,20 @@ def create_app(settings: Settings | None = None) -> FastAPI: """App factory: lifespan, routers, exception handler, /metrics.""" settings = settings or load_settings() - # FIRST, before any router is built and before any logger is bound. Without this the - # API is the one service of three that never configures structlog: its lines go out - # through logging.lastResort as bare text on stderr with no service, no trace_id and - # no JSON envelope — a parse failure in the collector, and unattributable in Loki. + # FIRST, before any router is built and any logger is bound. Without it the API is the + # one service of three that never configures structlog, and its lines go out through + # logging.lastResort as bare text on stderr — no service, no trace_id, no JSON envelope. # `settings.log_json` was silently inert here for the same reason. obs.setup("svcforge-api", settings) # Refuse the dev escape hatches when SVCFORGE_ENVIRONMENT says this is not a laptop. - # Called unconditionally and early: a check that only runs from a branch someone - # remembered to write is a check that does not run. + # Unconditional and early: a check that runs only from a branch someone remembered to + # write is a check that does not run. settings.check_production() - # The description is the API's documentation. FastAPI renders it as markdown at /docs, - # and it is the only place a caller who does not have this repo can learn the two things - # that are not obvious from the schema: every write is asynchronous, and the instance - # lifecycle is a state machine they have to poll. + # The description is the API's documentation, rendered as markdown at /docs. It is the + # only place a caller without this repo learns the two things the schema cannot say: + # every write is asynchronous, and the lifecycle is a state machine they have to poll. app = FastAPI( title="svcforge", version="0.1.0", @@ -227,7 +219,6 @@ def app() -> FastAPI: return create_app() -# There is deliberately no `if __name__ == "__main__"` here. `services/api/__main__.py` is -# the single entrypoint, and the image's ENTRYPOINT uses it. A second one in this module -# drifted from it — different log_level, different access_log — so `python -m services.api` -# and `python services/api/main.py` started the same app two different ways. +# No `if __name__ == "__main__"` here on purpose. `services/api/__main__.py` is the single +# entrypoint and the image's ENTRYPOINT uses it. A second one in this module drifted from +# it — different log_level, different access_log — so the same app started two ways. diff --git a/services/api/models.py b/services/api/models.py index a4c0b8a..e708dfb 100644 --- a/services/api/models.py +++ b/services/api/models.py @@ -1,9 +1,8 @@ -"""Wire types. +"""Wire types, deliberately not the domain models. -These are deliberately NOT the domain models. `Instance` carries `team`, `namespace` and -`release_name` — placement details a tenant has no business seeing and no business -setting. The response model is the allow-list that keeps them off the wire, which is why -it is written out by hand instead of derived from `Instance`. +`Instance` carries `team`, `namespace` and `release_name` — placement details a tenant has +no business seeing or setting. The response model is the allow-list that keeps them off the +wire, which is why it is written by hand instead of derived from `Instance`. """ from __future__ import annotations @@ -19,9 +18,8 @@ class CreateInstanceRequest(BaseModel): """What a tenant may ask for. `service_type` and `size` are plain strings, not enums: the catalog is data loaded at - runtime, so baking its keys into a type would mean a redeploy to add a service type, - and a 422 (schema) where the spec wants a 404 (unknown resource). They are validated - against the catalog in the handler. + runtime, so baking its keys into a type would mean a redeploy to add a service type and + a 422 where the spec wants a 404. The handler validates them against the catalog. """ model_config = ConfigDict( diff --git a/services/api/routes/health.py b/services/api/routes/health.py index 49d9e8e..5ac5f92 100644 --- a/services/api/routes/health.py +++ b/services/api/routes/health.py @@ -3,11 +3,11 @@ The distinction between the first two is the difference between a 30-second blip and a fleet-wide outage: -* `/healthz` (liveness) answers "is this process wedged?" A failure here gets the - container KILLED. It must therefore touch NOTHING external. Wire it to the DB and a - 20-second Postgres failover restarts every pod at once; they come back, find the DB - still down, and CrashLoopBackOff with exponential restart delays — so the fleet is now - down for minutes after the database recovered. +* `/healthz` (liveness) answers "is this process wedged?" A failure here KILLS the + container, so it must touch nothing external. Wired to the DB, a 20-second Postgres + failover restarts every pod at once; they come back, find the DB still down, and + CrashLoopBackOff with exponential delays — the fleet stays down for minutes after the + database recovered. * `/readyz` (readiness) answers "should this pod get traffic?" A failure here only removes it from the Service endpoints. It is allowed to check dependencies, and it recovers by itself the moment the check passes. @@ -26,10 +26,10 @@ from services.api.models import ErrorBody router = APIRouter(tags=["ops"]) -# No PROMETHEUS_MULTIPROC_DIR here, deliberately: it exists for prefork servers where each -# worker process holds a slice of the counters. One uvicorn process per container means -# the default in-process registry is already correct, and multiproc mode would add a -# shared temp dir, a cleanup obligation, and a class of stale-file bugs for nothing. +# No PROMETHEUS_MULTIPROC_DIR, deliberately: it exists for prefork servers where each +# process holds a slice of the counters. One uvicorn process per container makes the +# in-process registry correct, and multiproc mode would add a shared temp dir, a cleanup +# obligation, and a class of stale-file bugs for nothing. @router.get("/healthz", status_code=status.HTTP_200_OK) @@ -45,10 +45,9 @@ async def healthz() -> dict[str, str]: async def readyz(pool: PoolDep) -> dict[str, str]: """Readiness. Postgres only. - Postgres-only is the rule, and Redis is the temptation. Redis holds derived state — - rate-limit buckets, caches — and everything degrades gracefully without it. Put it in - this check and an Upstash hiccup marks every pod unready, Kubernetes empties the - Service, and a cache outage becomes a total API outage. + Redis is the temptation and stays out: it holds derived state that degrades gracefully, + so checking it here would let an Upstash hiccup mark every pod unready, empty the + Service, and turn a cache outage into a total API outage. """ try: async with pool.connection() as conn, conn.cursor() as cur: @@ -68,14 +67,13 @@ async def readyz(pool: PoolDep) -> dict[str, str]: async def metrics(request: Request) -> Response: """The Prometheus scrape endpoint. - A route rather than `app.mount("/metrics", make_asgi_app())`, for two reasons. A - Starlette `Mount` compiles to `^/metrics(?P/.*)$`, which does not match a bare - `/metrics` — the exact URL every scrape config uses — and a `Mount` is invisible to - OpenAPI, while the deliverable asks for `/metrics` in `openapi.json`. + A route rather than `app.mount("/metrics", make_asgi_app())`: a Starlette `Mount` + compiles to `^/metrics(?P/.*)$`, which does not match the bare `/metrics` every + scrape config uses, and a Mount is invisible to OpenAPI. - The encoding is still prometheus_client's: `choose_encoder` reads the Accept header and - picks the exposition format (Prometheus text vs OpenMetrics) with its matching content - type. Hand-rolling either is how you end up serving text/plain that a scraper rejects. + The encoding stays prometheus_client's — `choose_encoder` reads Accept and picks the + exposition format with its matching content type. Hand-rolling it serves text/plain a + scraper rejects. """ encoder, content_type = choose_encoder(request.headers.get("Accept", "")) return Response(content=encoder(REGISTRY), media_type=content_type) diff --git a/services/api/routes/instances.py b/services/api/routes/instances.py index 7f4c54a..cf401ae 100644 --- a/services/api/routes/instances.py +++ b/services/api/routes/instances.py @@ -1,12 +1,12 @@ """The tenant-facing API. -Two rules run through every handler here: +Two rules run through every handler: -* **AuthZ is the WHERE clause.** No handler ever compares `inst.team` to the caller's - team, because the repo never returns another team's row to compare. A wrong-team id is - a 404. 403 would confirm the id exists, which is the leak. -* **The instance and its task commit together.** A committed instance with no task is an - instance that never provisions and that nothing will ever retry. +* **AuthZ is the WHERE clause.** No handler compares `inst.team` to the caller's team, + because the repo never returns another team's row to compare. A wrong-team id is a 404; + 403 would confirm the id exists. +* **The instance and its task commit together.** A committed instance with no task never + provisions and nothing retries it. """ from __future__ import annotations @@ -30,9 +30,9 @@ from services.api.models import CreateInstanceRequest, ErrorBody, InstanceRespon from svcforge_core.domain.models import CatalogEntry, Instance, TaskKind from svcforge_core.domain.states import IllegalTransition, InstanceState, transition -# Declared on the router so every error shape lands in openapi.json under ErrorBody. -# The exception handler already renders this at runtime; without declaring it, generated -# clients see the contract for 2xx only and invent their own guess for the rest. +# Declared on the router so every error shape lands in openapi.json under ErrorBody. The +# exception handler already renders these at runtime; undeclared, a generated client sees +# the contract for 2xx only and guesses the rest. ERROR_RESPONSES: dict[int | str, dict[str, Any]] = { 401: {"model": ErrorBody, "description": "Missing or invalid credentials"}, 404: {"model": ErrorBody, "description": "No such instance, or not this team's"}, @@ -46,14 +46,12 @@ router = APIRouter(prefix="/v1/instances", tags=["instances"], responses=ERROR_R def release_name_for(team: str, service_type: str, instance_id: UUID) -> str: """The helm release name. Deterministic, and `unique` in the schema. - This is the idempotency anchor. A worker that dies after `helm install` but before it - marks the task done will retry, compute the same name, and `helm upgrade --install` - onto the same release instead of creating a second one. Derive it from anything that - is not already durable — a timestamp, a random suffix, the retry count — and a retry - provisions a duplicate. + The idempotency anchor: a worker that dies after `helm install` but before marking the + task done retries, computes the same name, and upgrades the same release instead of + creating a second one. Derive it from anything not already durable — a timestamp, a + random suffix, the retry count — and a retry provisions a duplicate. - Truncated to the uuid's first 8 chars to stay inside the 53-char limit helm imposes - on release names (Kubernetes label values, minus room for chart-generated suffixes). + Truncated to the uuid's first 8 chars to stay inside helm's 53-char release-name limit. """ return f"{team}-{service_type}-{str(instance_id)[:8]}" @@ -66,9 +64,8 @@ def namespace_for(team: str) -> str: def _resolve(catalog: dict[str, CatalogEntry], service_type: str, size: str) -> CatalogEntry: """Look up service_type + size, or raise the right 4xx. - The two failures are different HTTP problems and the spec asks for different codes: - an unknown service_type is a resource that does not exist (404); an unknown size for a - real service_type is a body the server understood and cannot process (422). + Two different HTTP problems: an unknown service_type is a resource that does not exist + (404), an unknown size for a real one is a body understood and unprocessable (422). """ entry = catalog.get(service_type) if entry is None: @@ -107,9 +104,9 @@ async def create_instance( ) -> Instance: """Accept a provisioning request. 202, never 201. - Nothing is provisioned when this returns. The row exists and a task is queued; a - worker will do the work seconds or minutes from now. 201 Created would be a lie about - a resource that does not exist yet, and clients would stop polling. + Nothing is provisioned when this returns: the row exists and a task is queued, and a + worker does the work seconds or minutes later. 201 Created would be a lie about a + resource that does not exist yet, and clients would stop polling. """ entry = _resolve(catalog, body.service_type, body.size) @@ -123,9 +120,9 @@ async def create_instance( state=InstanceState.REQUESTED, namespace=namespace_for(team), release_name=release_name_for(team, body.service_type, instance_id), - # Pinned from the catalog AT CREATION TIME, not read from the catalog later. - # This column records what is actually deployed; bumping catalog.yaml must show up - # as drift the reconciler can see, not silently rewrite history. + # Pinned at creation time, not read from the catalog later. The column records what + # is deployed, so bumping catalog.yaml shows up as drift the reconciler can see + # rather than silently rewriting history. chart_version=entry.chart_version, expires_at=now + timedelta(days=body.ttl_days) if body.ttl_days is not None else None, created_at=now, @@ -181,12 +178,11 @@ async def delete_instance( ) -> Instance: """state -> deleting, enqueue deprovision. 202: the helm uninstall has not happened yet. - Ordering note. `InstanceRepo.update_state` owns its own connection, so the CAS and the - enqueue cannot share one transaction without reaching around the repo. Given two - statements, the order is chosen for its failure mode: CAS first, enqueue second. A - crash in between leaves an instance in `deleting` with no task, which the reconciler's - sweep re-enqueues. The other order leaves a deprovision task pointing at a `ready` - instance, and a worker would tear down a live service nobody asked to delete. + `InstanceRepo.update_state` owns its own connection, so the CAS and the enqueue cannot + share a transaction without reaching around the repo. Given two statements, the order is + chosen for its failure mode: a crash between CAS and enqueue leaves an instance in + `deleting` with no task, which the reconciler's sweep re-enqueues. The reverse would + leave a deprovision task on a `ready` instance and tear down a live service. """ inst = await instances.get(instance_id, team) if inst is None: diff --git a/services/cli/main.py b/services/cli/main.py index c238a03..b488c4d 100644 --- a/services/cli/main.py +++ b/services/cli/main.py @@ -1,9 +1,9 @@ """svcforge — the control plane client. -This talks to the API over HTTP and never touches the database. That restraint is the -whole design: if the CLI could write to Postgres, every invariant the API enforces -(the state machine, the one-transaction create, AuthZ in the WHERE clause) would have a -back door, and the first 3am incident would go through it. +Talks to the API over HTTP and never touches the database. If the CLI could write to +Postgres, every invariant the API enforces — the state machine, the one-transaction create, +AuthZ in the WHERE clause — would have a back door, and the first 3am incident would go +through it. `ClientSettings` below is what keeps that true in practice. """ from __future__ import annotations @@ -16,9 +16,9 @@ from typing import Annotated, Any import httpx import typer +from pydantic_settings import BaseSettings, SettingsConfigDict from svcforge_core.domain.states import InstanceState -from svcforge_core.settings import load_settings app = typer.Typer(help="svcforge control plane client", no_args_is_help=True) @@ -41,8 +41,24 @@ class Size(StrEnum): MEDIUM = "medium" +class ClientSettings(BaseSettings): + """The two values the CLI needs, and nothing else. + + Its own model rather than `svcforge_core.settings.Settings`, which requires + `SVCFORGE_PG_DSN`: loading that here would refuse to run the CLI without a database URL + it then never opens, on a laptop that has no reason to hold one. + """ + + model_config = SettingsConfigDict( + env_prefix="SVCFORGE_", env_file=".env", env_file_encoding="utf-8", extra="ignore", frozen=True + ) + + api_url: str = "http://localhost:8000" + api_token: str | None = None + + def _client() -> httpx.Client: - settings = load_settings() + settings = ClientSettings() headers = {"authorization": f"Bearer {settings.api_token}"} if settings.api_token else {} return httpx.Client(base_url=settings.api_url, headers=headers, timeout=10.0) diff --git a/services/reconciler/main.py b/services/reconciler/main.py index 8a94620..c98fb7e 100644 --- a/services/reconciler/main.py +++ b/services/reconciler/main.py @@ -1,29 +1,28 @@ """The control loop. -Every other service in svcforge is edge-triggered: a tenant POSTs, a row appears, a worker -claims it. Edge-triggered systems are correct exactly as long as nothing is ever missed — -and things are missed. A worker is SIGKILLed holding a lease. An operator runs -`helm uninstall` by hand. A pod dies between the CAS and the enqueue. Nobody sends an event -for any of that, because the thing that would have sent it is the thing that died. +Every other service here is edge-triggered: a tenant POSTs, a row appears, a worker claims +it. That is correct only as long as nothing is missed, and things are missed — a worker +SIGKILLed holding a lease, an operator running `helm uninstall` by hand, a pod dying +between the CAS and the enqueue. Nothing sends an event for any of it, because the thing +that would have sent it is the thing that died. So: level-triggered. Every 60 seconds, compare the world to the database and enqueue what -is missing. The four checks below do not know or care what went wrong, or whether anything -did; they are the same code on the happy path and after an outage. That property is the -entire reason this service exists, and it is why each check is written as a *query for -work*, never as a reaction to an event. +is missing. The four checks below do not know what went wrong, or whether anything did; +they are the same code on the happy path and after an outage. That is why each is written +as a query for work rather than a reaction to an event. Three rules hold the design together: -* **Singleton.** `replicas: 1`, `strategy: Recreate` in the chart. Two reconcilers - double-enqueue drift and race on TTL. There is no leader election here on purpose: the - correct lease for that lives in Postgres next to the data, not in a Redis lock, and - until there is a second replica to elect between, an election is a subsystem that can - only fail. One pod, and the `SvcforgeReconcilerStale` alert is what notices it is gone. -* **Each check is independent.** One failing check must not skip the other three. A helm - 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 lists the live releases, because seeing reality is the job. +* **Singleton.** `replicas: 1`, `strategy: Recreate`. Two reconcilers double-enqueue drift + and race on TTL. No leader election on purpose — the right lease for that lives in + Postgres next to the data, and until there is a second replica to elect between, an + election is a subsystem that can only fail. The `SvcforgeReconcilerStale` alert notices + when the one pod is gone. +* **Each check is independent.** A helm binary that cannot reach the API server must not + stop TTLs from expiring. +* **Enqueue, never act.** The reconciler diagnoses and workers treat: it writes task rows + and instance states and never calls `helm install`. Reading is the exception, since + seeing reality is the job. """ from __future__ import annotations @@ -63,9 +62,9 @@ log = get_logger("svcforge.reconciler") class ReconcilerDeps: """Everything a check is allowed to touch. Built once in main(), passed down. - Same shape as `WorkerDeps` for the same reason: the checks take `deps` instead of - reaching for globals, so the integration tests below run every check against a real - Postgres and a `FakeProvisioner` without a cluster anywhere in sight. + Same shape as `WorkerDeps` and for the same reason: checks take `deps` instead of + reaching for globals, so the integration tests run every check against a real Postgres + and a `FakeProvisioner` with no cluster in sight. """ pool: DictPool @@ -90,21 +89,17 @@ class ReconcilerDeps: async def check_drift(deps: ReconcilerDeps) -> None: """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 - and the release never came back. The DB still says `ready` and still hands the tenant an - endpoint that resolves to nothing. + The only check that looks outside Postgres, and the only one that catches someone + running `helm uninstall` by hand or a drained node whose release never came back — where + the DB still says `ready` and still hands the tenant an endpoint resolving to nothing. - Two directions, two very different answers: + Two directions, two different answers: - * **Release gone, DB says `ready`** -> re-enqueue provision. Safe, because provisioning - is `helm upgrade --install` against a deterministic release name: converging on - desired state, not a blind re-install. - * **Release exists, DB knows nothing** -> log at error with release and namespace, and - stop. **Never delete in v1.** The reconciler's view of "the DB knows nothing" is one - query against one database; the release might belong to another team, another tool, - or a migration half-finished. Deleting on that evidence is how an automated system - takes down production faster than any human could. A human reads the log and decides. + * **Release gone, DB says `ready`** -> re-enqueue provision. Safe because provisioning + is `helm upgrade --install` against a deterministic release name. + * **Release exists, DB knows nothing** -> log at error and stop. **Never delete in v1.** + "The DB knows nothing" is one query against one database, and the release might belong + to another team, another tool, or a half-finished migration. A human decides. """ with tracer().start_as_current_span("helm.list"): releases = await deps.provisioner.list_releases() @@ -132,31 +127,30 @@ async def check_drift(deps: ReconcilerDeps) -> None: {"instance_id": str(inst.id), "team": inst.team}, ) except Exception: - # The task is already committed; the notification is a courtesy. A webhook - # timing out must not abandon the rest of the sweep — the instances after this - # one in the loop have the same problem and nobody else is coming to find them. + # The task is committed; the notification is a courtesy. A webhook timing out + # must not abandon the rest of the sweep — the instances after this one have the + # same problem and nobody else is coming to find them. log.exception("notify.failed", instance_id=str(inst.id)) known = await deps.reconcile.known_releases() for name, namespace in sorted(live - known): - # error, not warning: this is a resource nobody is billing for and nobody owns. - # It will sit here every 60s until a human deletes it or adopts it. That is the - # intended pressure. + # error, not warning: a resource nobody owns and nobody is billing for. It repeats + # every 60s until a human deletes or adopts it, which is the intended pressure. log.error("drift.orphan_release", release=name, namespace=namespace, action="none (v1 never deletes)") async def check_lease_expiry(deps: ReconcilerDeps) -> None: """Tasks whose worker died -> back to `queued`. - A lease. No lock survives a power cut: a worker SIGKILLed mid-provision - leaves `state='running'` with `locked_by` set and nobody running it, and no amount of - cleanup code in the worker helps, because the worker is the part that died. `locked_at` - plus a timeout is the only thing that recovers the row, which is why `locked_at` exists. + A lease, not a lock: no lock survives a power cut. A worker SIGKILLed mid-provision + leaves `state='running'` with `locked_by` set and nobody running it, and cleanup code in + the worker cannot help because the worker is what died. `locked_at` plus a timeout is + the only thing that recovers the row. - The 5-minute default must exceed the longest a healthy task can hold a lease, or the - reconciler hands a still-running provision to a second worker. Handlers are idempotent, - so that is survivable, though it still costs a duplicated helm run — which is why - `lease_seconds` sits above helm's `--timeout`. + The 5-minute default must exceed the longest a healthy task can hold a lease, or a + still-running provision is handed to a second worker. Handlers are idempotent so that is + survivable, but it costs a duplicated helm run — hence `lease_seconds` > helm's + `--timeout`. """ freed = await deps.tasks.reset_expired_leases(deps.settings.lease_seconds) if freed: @@ -166,14 +160,13 @@ async def check_lease_expiry(deps: ReconcilerDeps) -> None: async def check_ttl(deps: ReconcilerDeps) -> None: """Expired instances -> `deleting`, plus a deprovision task. - The line item that stops a demo cluster from becoming a permanent cloud bill. Also the - sweep the API's DELETE route depends on: it CASes to `deleting` and enqueues in two - statements, and a crash in between lands here on the next tick. + What stops a demo cluster becoming a permanent cloud bill, and the sweep the API's + DELETE route depends on: DELETE CASes and enqueues in two statements, and a crash + between them lands here on the next tick. - Idempotent by construction — the work list excludes anything that already has a queued - or running deprovision, and the CAS and the insert share one transaction. Without that - guard, a deprovision that takes longer than 60 seconds gets a second task on the next - tick, and a third on the tick after. + Idempotent by construction — the work list excludes anything with a queued or running + deprovision, and the CAS and insert share one transaction. Without that, a deprovision + taking longer than 60 seconds collects a new task every tick. """ for inst in await deps.reconcile.due_for_deprovision(): task_id = await deps.reconcile.enqueue_deprovision(inst.id) @@ -192,20 +185,18 @@ async def check_ttl(deps: ReconcilerDeps) -> None: async def check_version_drift(deps: ReconcilerDeps) -> None: """The day-2 rollout: the work-list query, one service type at a time. - Everything that makes this safe is somewhere else, which is the point: + Everything that makes it safe is somewhere else, which is the point: * `list_upgradable` limits to `max_in_flight` and returns nothing while `rollout_state='halted'`, so a bad chart stops after one tenant. - * `schedule_upgrade_at` turns the tenant's maintenance window into a `run_after`; the - queue does the waiting, in `where run_after <= now()`. There is no scheduler here and - there must not be one — a task parked in Postgres until 03:00 Sunday survives a - reconciler restart, and an in-memory timer does not. - * `security: true` in the catalog bypasses the window. A CVE with a public exploit does - not wait until Sunday. + * `schedule_upgrade_at` turns the maintenance window into a `run_after` and the queue + does the waiting, in `where run_after <= now()`. No scheduler here, and there must not + be one: a task parked in Postgres until 03:00 Sunday survives a restart, a timer does + not. + * `security: true` in the catalog bypasses the window. - A bad window spec is this instance's problem, not the fleet's: log it and move to the - next one. Failing the whole check would let one tenant's typo freeze everyone's - security rollout. + A bad window spec is one instance's problem: log it and move on. Failing the check would + let one tenant's typo freeze everyone's security rollout. """ now = deps.clock.now() @@ -260,23 +251,20 @@ CHECKS: dict[str, Callable[[ReconcilerDeps], Awaitable[None]]] = { async def tick(deps: ReconcilerDeps) -> None: """One pass: all four checks, then the gauges, then the heartbeat. - Checks first, gauges second: `svcforge_queue_depth` is read straight after the checks - that add to the queue, so the value scraped is the value the tick left behind rather - than one from before its own work. + Checks first, gauges second, so `svcforge_queue_depth` reports what this tick left + behind rather than what preceded its own work. - The heartbeat is set unconditionally, and that is deliberate. It answers "is the loop - running", not "is everything fine" — the checks have their own alerts. Gating it on - success would make `SvcforgeReconcilerStale` fire for a helm blip and mean two things - at once, and an alert that means two things gets muted. + The heartbeat is set unconditionally. It answers "is the loop running", not "is + everything fine" — the checks have their own alerts. Gating it on success would make + `SvcforgeReconcilerStale` fire for a helm blip and mean two things at once, and an alert + that means two things gets muted. - The whole tick runs inside one span, which is a considered exception to "manual spans go - around helm calls only". That rule exists so the API does not hand-roll spans that - `opentelemetry-instrument` already creates for it. Nothing auto-instruments the - reconciler: without a span here it emits no traces at all, and — because - `inject_traceparent` serialises the *active* context — every task it enqueues would be - written with a null `traceparent` and be unjoinable to the tick that decided to create - it. One span per tick is what makes "why was this instance re-provisioned at 03:00?" a - question the traces can answer. + The whole tick runs in one span, a considered exception to "manual spans wrap helm calls + only". That rule keeps the API from hand-rolling spans `opentelemetry-instrument` + already makes; nothing auto-instruments the reconciler, so without this it emits no + traces at all and — since `inject_traceparent` serialises the *active* context — every + task it enqueues would carry a null `traceparent` and be unjoinable to the tick that + created it. """ with tracer().start_as_current_span("reconciler.tick"): await _run_checks(deps) @@ -291,12 +279,11 @@ async def _run_checks(deps: ReconcilerDeps) -> None: try: await check(deps) except Exception: # the tick is the error boundary - # The swallow is the design. These four checks share nothing but a database - # handle, and the value of a level-triggered loop is that it keeps running: an - # unreachable cluster must not stop TTLs from expiring, and one tenant's broken + # The swallow is the design. The four checks share nothing but a database + # handle, and a level-triggered loop is only worth having if it keeps running: + # an unreachable cluster must not stop TTLs expiring, and one tenant's broken # window spec must not stop drift detection. This means "this check achieved - # nothing for 60 seconds", which the log says out loud. It never means "the - # reconciler stops". + # nothing for 60 seconds", never "the reconciler stops". log.exception("check.failed", check=name) try: @@ -311,10 +298,10 @@ async def _run_checks(deps: ReconcilerDeps) -> None: async def run_reconciler(deps: ReconcilerDeps, stop: asyncio.Event) -> None: """Tick, sleep, repeat, until told to stop. - Tick first, then sleep: a pod that has just been restarted should reconcile now, not in - sixty seconds. Fixed interval rather than a fixed period — a tick that overruns simply - delays the next one, instead of stacking a second tick on top of the first, which for a - singleton would be exactly the concurrent reconciler `replicas: 1` exists to prevent. + Tick first, then sleep: a just-restarted pod should reconcile now, not in sixty seconds. + Fixed interval rather than fixed period, so a tick that overruns delays the next one + instead of stacking a second on top — which for a singleton is exactly the concurrent + reconciler `replicas: 1` exists to prevent. """ while not stop.is_set(): await tick(deps) @@ -353,13 +340,13 @@ async def _amain(once: bool, own_team: str, max_in_flight: int) -> None: try: if once: - # One pass and exit: the acceptance path, and how you drive a reconcile by hand - # from a shell. No metrics server — nothing would ever scrape it. + # One pass and exit: the acceptance path, and how to drive a reconcile by hand. + # No metrics server — nothing would ever scrape it. await tick(deps) return - # 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. + # settings.metrics_port, like the worker. SVCFORGE_METRICS_PORT 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() diff --git a/services/worker/handlers.py b/services/worker/handlers.py index e8f78d3..8cb7053 100644 --- a/services/worker/handlers.py +++ b/services/worker/handlers.py @@ -1,13 +1,12 @@ """Task handlers. -Every handler here obeys one rule: running it twice must equal running it once. +Every handler obeys one rule: running it twice must equal running it once. -A worker can be SIGKILLed after helm has installed the release but -before the DB row says so; the lease expires; another worker claims the same task and runs -this function again. If the handler is not idempotent, the tenant gets two Elasticsearches -and you get a bill. Idempotency is what makes the crash safe, and it is bought in two -places: a deterministic `release_name`, and adapters that state desired state -(`helm upgrade --install`) instead of issuing imperative commands. +A worker can be SIGKILLed after helm installed the release but before the DB row says so; +the lease expires, another worker claims the same task, and this function runs again. A +handler that is not idempotent gives the tenant two Elasticsearches and you a bill. +Idempotency is bought in two places: a deterministic `release_name`, and adapters that +state desired state (`helm upgrade --install`) instead of issuing imperative commands. """ from __future__ import annotations @@ -85,11 +84,10 @@ async def handle_provision(task: Task, deps: WorkerDeps) -> None: {"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. + # The provision succeeded and the row is READY; the notification is a courtesy. + # Propagating a webhook timeout 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. log.exception("notify.failed", instance_id=str(inst.id)) @@ -104,10 +102,9 @@ async def handle_deprovision(task: Task, deps: WorkerDeps) -> None: # swallows not-found, because the desired state — no release — is already true. await deps.provisioner.uninstall(release=inst.release_name, ns=inst.namespace) - # Raise rather than ignore the CAS result. Swallowing it means: the release is gone, - # the row keeps `state=ready` and its now-dangling endpoint, the task is marked done, - # and 60 seconds later the reconciler's drift check re-provisions the thing the tenant - # asked to delete. Failing loudly turns a silent ping-pong into one visible error. + # Raise rather than ignore the CAS result. Swallowing it leaves the release gone, the + # row on `state=ready` with a dangling endpoint, the task marked done — and 60 seconds + # later the drift check re-provisions the thing the tenant asked to delete. if not await deps.instances.update_state(inst.id, InstanceState.DELETING, InstanceState.DELETED): raise HandlerError( f"instance {inst.id} was {inst.state.value}, expected {InstanceState.DELETING.value}" @@ -146,10 +143,9 @@ async def handle_upgrade(task: Task, deps: WorkerDeps) -> None: async def handle_verify(task: Task, deps: WorkerDeps) -> None: """Post-upgrade health probe. On failure, halt the whole rollout for this service type. - One column decides whether the fleet keeps rolling. The work-list query returns nothing - while `rollout_state='halted'`, so a bad chart stops after the first tenant instead of - after all of them. You clear it with SQL, deliberately: an automatic un-halt would just - resume breaking things. + The work-list query returns nothing while `rollout_state='halted'`, so a bad chart stops + after the first tenant instead of all of them. Clearing it is a deliberate SQL statement: + an automatic un-halt would resume breaking things. """ inst = await _load_instance(task, deps) releases = {r.name for r in await deps.provisioner.list_releases()} @@ -157,10 +153,9 @@ async def handle_verify(task: Task, deps: WorkerDeps) -> None: if inst.release_name in releases: return - # `returning` + a `where` on the update half tells us whether THIS call was the one - # that halted the rollout. The halt itself is idempotent; the page is not. Without the - # distinction, a verify that fails its full retry budget sends five identical - # notifications for one incident, spread across the backoff curve. + # `returning` plus a `where` on the update half says whether THIS call halted the + # rollout. The halt is idempotent; the page is not. Without the distinction, a verify + # that burns its full retry budget sends five identical notifications for one incident. async with deps.pool.connection() as conn, conn.cursor() as cur: await cur.execute( """insert into catalog_versions (service_type, rollout_state) diff --git a/services/worker/main.py b/services/worker/main.py index 1733627..8077a48 100644 --- a/services/worker/main.py +++ b/services/worker/main.py @@ -1,10 +1,9 @@ """The claim loop. -Poll every 5 seconds. Claim while a semaphore slot is free. Run the handler. Report. -That is the whole design, and the restraint is the point: LISTEN/NOTIFY would shave the -latency, is fire-and-forget so it can never replace the poll anyway, is strictly extra -code, and does not exist on pgbouncer's transaction pooler. The poll is not a placeholder -for something better. +Poll every 5 seconds. Claim while a semaphore slot is free. Run the handler. Report. The +poll is not a placeholder for something better: LISTEN/NOTIFY would shave latency, but it +is fire-and-forget so it can never replace the poll, and it does not exist on pgbouncer's +transaction pooler. """ from __future__ import annotations @@ -35,17 +34,16 @@ log = obs.get_logger("svcforge.worker") async def _report(coro: Awaitable[bool], task_id: int, what: str) -> None: """Run a terminal report, and never let its failure escape. - Reporting is the one thing that must not kill the worker. `_run_one` runs inside a - TaskGroup, and a TaskGroup cancels every sibling the moment one child raises — so a - DB blip during `tasks.fail()` would abort every other in-flight provision on this pod, - not just this one. The task itself is safe either way: it stays `running` and the - reconciler's lease sweep returns it to the queue. Losing the report costs one lease - interval; losing the siblings costs their work. + `_run_one` runs inside a TaskGroup, which cancels every sibling the moment one child + raises — so a DB blip during `tasks.fail()` would abort every other in-flight provision + on this pod. The task itself is safe either way: it stays `running` and the lease sweep + returns it to the queue. Losing the report costs one lease interval; losing the siblings + costs their work. """ try: if not await coro: - # The lease was stolen while we were working: another worker owns this task - # now and is mid-run. Reporting is theirs to do, not ours. + # The lease was stolen while we were working: another worker owns this task now + # and is mid-run. Reporting is theirs, not ours. log.warning("lease lost before report; another worker owns this task", task_id=task_id) except Exception: log.exception("could not report task %s (%s); lease will expire", task_id, what) @@ -55,10 +53,8 @@ async def _run_one(deps: WorkerDeps, task: Task, sem: asyncio.Semaphore) -> None """Run one task to a terminal report. Never lets an exception escape the TaskGroup.""" worker_id = deps.settings.worker_id try: - # Every log line from here carries instance_id/task_id/team. Bound once, at claim, - # rather than passed down: the alternative is threading three arguments through - # every function that might log, and the first one anyone forgets is the one you - # need at 3am. + # Every log line from here carries instance_id/task_id/team. Bound once at claim + # rather than threaded through every function that might log. obs.bind_task_context(task.instance_id, task.id, team=task.team or "unknown") log.info("task claimed", kind=task.kind.value, attempt=task.attempts) obs.TASKS_CLAIMED.labels(kind=task.kind.value).inc() @@ -72,9 +68,9 @@ async def _run_one(deps: WorkerDeps, task: Task, sem: asyncio.Semaphore) -> None ) return - # Re-parent to the span that enqueued this task. Without the stored traceparent - # the worker's span starts a brand-new trace, and the POST that caused the work - # is in a different trace to the helm call that did it. + # Re-parent to the span that enqueued this task. Without the stored traceparent the + # worker's span starts a new trace, putting the POST that caused the work in a + # different trace from the helm call that did it. ctx = obs.context_from_traceparent(task.traceparent) started = time.monotonic() with obs.tracer().start_as_current_span( @@ -108,10 +104,9 @@ async def _run_one(deps: WorkerDeps, task: Task, sem: asyncio.Semaphore) -> None "fail", ) else: - # Only provisions go in the provision histogram. The buckets run 10s..1800s - # because they were sized for helm installs; a sub-second `verify` dropped - # into the same series drags the p95 down and quietly stops - # SvcforgeProvisionSlow from ever firing. + # Only provisions go in the provision histogram. Its buckets run 10s..1800s + # for helm installs, so a sub-second `verify` in the same series drags the + # p95 down and quietly stops SvcforgeProvisionSlow from ever firing. if task.kind is TaskKind.PROVISION: obs.PROVISION_TIME.observe(time.monotonic() - started) await _report(deps.tasks.complete(task.id, worker_id), task.id, "complete") @@ -122,10 +117,9 @@ async def _run_one(deps: WorkerDeps, task: Task, sem: asyncio.Semaphore) -> None async def run_worker(deps: WorkerDeps, stop: asyncio.Event) -> None: """Claim and run until told to stop, then drain what is in flight. - Draining is what makes a rolling deploy invisible. Exiting the `async with` block - awaits every in-flight handler, so a pod that is being replaced finishes the provision - it already started instead of abandoning it half-done for the lease to clean up - five minutes later. + Draining is what makes a rolling deploy invisible: exiting the `async with` awaits every + in-flight handler, so a pod being replaced finishes the provision it started instead of + abandoning it for the lease to clean up five minutes later. """ sem = asyncio.Semaphore(deps.settings.worker_concurrency) worker_id = deps.settings.worker_id @@ -158,9 +152,8 @@ async def run_worker(deps: WorkerDeps, stop: asyncio.Event) -> None: async def _amain() -> None: settings: Settings = load_settings() - # Before anything else: nothing logged above this line is structured, and the metrics - # the SvcforgeTaskFailed / SvcforgeProvisionSlow alerts query do not exist until the - # registry is up. + # Before anything else: nothing above this line logs structured, and the metrics the + # SvcforgeTaskFailed / SvcforgeProvisionSlow alerts query do not exist until it runs. obs.setup("svcforge-worker", settings) settings.check_production() obs.start_metrics_server(settings.metrics_port)