fix(cd): bound post-rollout database verifier

.gitea/workflows/cd.yaml

@@ -3269,11 +3269,37 @@ jobs:
             container="$2"
             representative_probe_count="$3"
             required_connection_budget="$4"
-            $KUBECTL exec -i "deployment/$workload" -n awoooi-prod \
-              -c "$container" -- env \
-              AWOOOI_DB_REPRESENTATIVE_PROBE_COUNT="$representative_probe_count" \
-              AWOOOI_DB_REQUIRED_CONNECTION_BUDGET="$required_connection_budget" \
-              python - <<'PY'
+            last_failure="workload_not_ready"
+            ready_pod_ref=""
+            if ! $KUBECTL rollout status "deployment/$workload" \
+              -n awoooi-prod --timeout=20s >/dev/null 2>&1; then
+              last_failure="deployment_rollout_not_ready"
+            else
+              ready_pod_ref=$(
+                $KUBECTL get pods -n awoooi-prod -l "app=$workload" \
+                  --field-selector=status.phase=Running \
+                  --sort-by=.metadata.creationTimestamp \
+                  -o name 2>/dev/null | tail -n 1
+              ) || ready_pod_ref=""
+              if [ -z "$ready_pod_ref" ]; then
+                last_failure="running_pod_unavailable"
+              elif ! $KUBECTL wait --for=condition=Ready "$ready_pod_ref" \
+                -n awoooi-prod --timeout=10s >/dev/null 2>&1; then
+                last_failure="ready_pod_unavailable"
+              else
+                container_ready=$(
+                  $KUBECTL get "$ready_pod_ref" -n awoooi-prod \
+                    -o jsonpath="{.status.containerStatuses[?(@.name=='$container')].ready}" \
+                    2>/dev/null
+                ) || container_ready=""
+                if [ "$container_ready" != "true" ]; then
+                  last_failure="container_not_ready"
+                elif identity_output=$(
+                  $KUBECTL exec -i "$ready_pod_ref" -n awoooi-prod \
+                    -c "$container" -- env \
+                    AWOOOI_DB_REPRESENTATIVE_PROBE_COUNT="$representative_probe_count" \
+                    AWOOOI_DB_REQUIRED_CONNECTION_BUDGET="$required_connection_budget" \
+                    python - <<'PY'
           import asyncio
           import hashlib
           import json
@@ -3411,9 +3437,40 @@ jobs:
 
           asyncio.run(main())
           PY
+                ); then
+                  identity_json=$(printf '%s\n' "$identity_output" | tail -n 1)
+                  if printf '%s' "$identity_json" | python3 -c '
+          import json
+          import sys
+
+          payload = json.load(sys.stdin)
+          valid = (
+              bool(payload.get("role_fingerprint"))
+              and payload.get("representative_preflight_passed") is True
+              and int(
+                  payload.get("representative_connection_probe_count") or 0
+              ) >= 1
+          )
+          raise SystemExit(0 if valid else 1)
+          ' >/dev/null 2>&1; then
+                    printf '%s\n' "$identity_json"
+                    return 0
+                  fi
+                  last_failure="identity_preflight_not_ready"
+                else
+                  last_failure="kubectl_exec_failed"
+                fi
+              fi
+            fi
+            echo "workload_db_identity_once_failed workload=$workload container=$container reason=$last_failure" >&2
+            return 1
           }
           read_workload_db_identity() {
             workload="$1"
+            retry_backoff_seconds="${AWOOOI_DB_IDENTITY_RETRY_BACKOFF_SECONDS:-5}"
+            case "$retry_backoff_seconds" in
+              ''|*[!0-9]*) retry_backoff_seconds=5 ;;
+            esac
             attempt=1
             while [ "$attempt" -le 3 ]; do
               if identity_output=$(read_workload_db_identity_once "$@"); then
@@ -3422,11 +3479,12 @@ jobs:
               fi
               echo "workload_db_identity_exec_retry=$attempt/3;workload=$workload" >&2
               if [ "$attempt" -lt 3 ]; then
-                sleep $((attempt * 3))
+                sleep "$retry_backoff_seconds"
               fi
               attempt=$((attempt + 1))
             done
             echo "workload_db_identity_exec_exhausted=3;workload=$workload" >&2
+            printf '%s\n' '{"role_fingerprint":"","connection_limit":null,"required_connection_budget":null,"active_role_connection_count":null,"available_connection_headroom":null,"representative_connection_probe_count":0,"representative_preflight_passed":false,"table_privilege_gap_count":null,"sequence_privilege_gap_count":null,"error_type":"WorkloadDbIdentityAttemptsExhausted"}'
             return 1
           }
           json_field() {
@@ -3457,12 +3515,28 @@ jobs:
             fi
           }
 
-          API_DB_IDENTITY=$(read_workload_db_identity awoooi-api api 1 8 \
-            | tail -n 1 || true)
-          WORKER_DB_IDENTITY=$(read_workload_db_identity awoooi-worker worker 1 5 \
-            | tail -n 1 || true)
-          BROKER_DB_IDENTITY=$(read_workload_db_identity \
-            awoooi-ansible-executor-broker broker 1 2 | tail -n 1 || true)
+          if API_DB_IDENTITY=$(read_workload_db_identity awoooi-api api 1 8); then
+            API_DB_IDENTITY_EXIT=0
+          else
+            API_DB_IDENTITY_EXIT=$?
+          fi
+          if WORKER_DB_IDENTITY=$(read_workload_db_identity awoooi-worker worker 1 5); then
+            WORKER_DB_IDENTITY_EXIT=0
+          else
+            WORKER_DB_IDENTITY_EXIT=$?
+          fi
+          if BROKER_DB_IDENTITY=$(read_workload_db_identity \
+            awoooi-ansible-executor-broker broker 1 2); then
+            BROKER_DB_IDENTITY_EXIT=0
+          else
+            BROKER_DB_IDENTITY_EXIT=$?
+          fi
+          if [ "$API_DB_IDENTITY_EXIT" -ne 0 ] \
+            || [ "$WORKER_DB_IDENTITY_EXIT" -ne 0 ] \
+            || [ "$BROKER_DB_IDENTITY_EXIT" -ne 0 ]; then
+            echo "❌ workload DB identity attempts exhausted before receipt write"
+            exit 1
+          fi
           API_DB_ROLE_FINGERPRINT=$(printf '%s' "$API_DB_IDENTITY" \
             | json_field role_fingerprint 2>/dev/null || true)
           WORKER_DB_ROLE_FINGERPRINT=$(printf '%s' "$WORKER_DB_IDENTITY" \
@@ -3633,27 +3707,117 @@ jobs:
 
           API_HTTP_CONCURRENCY_VERIFIED=$(python3 - <<'PY'
           import concurrent.futures
+          import os
+          import sys
           import time
+          import urllib.error
           import urllib.request
 
           urls = (
-              "https://awoooi.wooo.work/api/v1/platform/events/dossier/coverage?project_id=awoooi",
-              "https://awoooi.wooo.work/api/v1/platform/cicd/events?project_id=awoooi",
-          ) * 6
+              # Keep the verifier representative and bounded: the connection
+              # concurrency is the subject under test, not a repeated scan of
+              # the default 100-row coverage window.
+              "https://awoooi.wooo.work/api/v1/platform/events/dossier/coverage?project_id=awoooi&limit=1",
+              "https://awoooi.wooo.work/api/v1/platform/cicd/events?project_id=awoooi&limit=1",
+          )
+          transient_http_statuses = {408, 425, 429, 500, 502, 503, 504}
+
+
+          def bounded_int(
+              name: str,
+              default: int,
+              minimum: int,
+              maximum: int,
+          ) -> int:
+              try:
+                  value = int(os.environ.get(name, str(default)))
+              except (TypeError, ValueError):
+                  value = default
+              return max(minimum, min(value, maximum))
+
+
+          readiness_attempts = bounded_int(
+              "API_PUBLIC_READINESS_ATTEMPTS", 12, 1, 20
+          )
+          readiness_delay_seconds = bounded_int(
+              "API_PUBLIC_READINESS_DELAY_SECONDS", 5, 0, 30
+          )
+          concurrency_attempts = bounded_int(
+              "API_HTTP_CONCURRENCY_ATTEMPTS", 4, 1, 8
+          )
+          concurrency_delay_seconds = bounded_int(
+              "API_HTTP_CONCURRENCY_DELAY_SECONDS", 5, 0, 30
+          )
+
+
+          def fetch_status(url: str, timeout: int) -> tuple[int | None, str]:
+              try:
+                  with urllib.request.urlopen(url, timeout=timeout) as response:
+                      return int(response.status), "response"
+              except urllib.error.HTTPError as exc:
+                  return int(exc.code), "HTTPError"
+              except Exception as exc:
+                  return None, type(exc).__name__
+
+
+          readiness_result = "unavailable"
+          for readiness_attempt in range(1, readiness_attempts + 1):
+              readiness_statuses = []
+              readiness_errors = []
+              for url in urls:
+                  status, error_type = fetch_status(url, timeout=10)
+                  readiness_statuses.append(status)
+                  readiness_errors.append(error_type)
+                  if (
+                      status is not None
+                      and status != 200
+                      and status not in transient_http_statuses
+                  ):
+                      raise SystemExit(
+                          "API sequential readiness permanent HTTP status: "
+                          + str(status)
+                      )
+              readiness_result = (
+                  "statuses="
+                  + ",".join(
+                      str(status) if status is not None else "unavailable"
+                      for status in readiness_statuses
+                  )
+                  + ";errors="
+                  + ",".join(readiness_errors)
+              )
+              if readiness_statuses and all(
+                  status == 200 for status in readiness_statuses
+              ):
+                  print(
+                      "api_sequential_readiness=ready;attempt="
+                      + str(readiness_attempt),
+                      file=sys.stderr,
+                  )
+                  break
+              if readiness_attempt < readiness_attempts:
+                  time.sleep(readiness_delay_seconds)
+          else:
+              raise SystemExit(
+                  "API sequential readiness failed: " + readiness_result
+              )
 
 
           def fetch(url: str) -> int:
-              with urllib.request.urlopen(url, timeout=20) as response:
-                  return int(response.status)
+              status, error_type = fetch_status(url, timeout=20)
+              if status is None:
+                  raise RuntimeError(error_type)
+              return status
 
 
           last_result = "unavailable"
-          for attempt in range(1, 4):
+          concurrency_urls = urls * 6
+          for attempt in range(1, concurrency_attempts + 1):
               try:
                   with concurrent.futures.ThreadPoolExecutor(
                       max_workers=8
                   ) as executor:
-                      statuses = list(executor.map(fetch, urls))
+                      statuses = list(executor.map(fetch, concurrency_urls))
               except Exception as exc:
                   last_result = type(exc).__name__
               else:
@@ -3661,8 +3825,22 @@ jobs:
                   if all(status == 200 for status in statuses):
                       print("true")
                       break
-              if attempt < 3:
-                  time.sleep(5)
+                  permanent_statuses = sorted({
+                      status
+                      for status in statuses
+                      if status != 200
+                      and status not in transient_http_statuses
+                  })
+                  if permanent_statuses:
+                      raise SystemExit(
+                          "API database concurrency probe permanent HTTP "
+                          "status: "
+                          + ",".join(
+                              str(status) for status in permanent_statuses
+                          )
+                      )
+              if attempt < concurrency_attempts:
+                  time.sleep(concurrency_delay_seconds)
           else:
               raise SystemExit(
                   "API database concurrency probe failed: " + last_result

apps/api/tests/test_workload_database_identity_projection.py

@@ -1,10 +1,137 @@
+import json
+import os
+import subprocess
+import textwrap
+import time
+import urllib.error
+import urllib.request
 from pathlib import Path
 
+import pytest
 import yaml
 
 REPO_ROOT = Path(__file__).resolve().parents[3]
 
 
+def _read_workload_db_identity_shell() -> str:
+    workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
+    start = workflow.index("          read_workload_db_identity_once() {")
+    end = workflow.index("          json_field() {", start)
+    return textwrap.dedent(workflow[start:end])
+
+
+def _api_http_readiness_and_concurrency_python() -> str:
+    workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
+    marker = "          API_HTTP_CONCURRENCY_VERIFIED=$(python3 - <<'PY'"
+    start = workflow.index(marker) + len(marker)
+    end = workflow.index("\n          PY\n          )", start)
+    return textwrap.dedent(workflow[start:end]).lstrip("\n")
+
+
+class _FakeHTTPResponse:
+    def __init__(self, status: int) -> None:
+        self.status = status
+
+    def __enter__(self) -> "_FakeHTTPResponse":
+        return self
+
+    def __exit__(self, *_args: object) -> None:
+        return None
+
+
+def _run_workload_db_identity_contract(
+    tmp_path: Path,
+    *,
+    exec_failures: int,
+) -> tuple[subprocess.CompletedProcess[str], dict[str, int]]:
+    state_dir = tmp_path / "state"
+    state_dir.mkdir()
+    fake_kubectl = tmp_path / "kubectl"
+    fake_kubectl.write_text(
+        textwrap.dedent(
+            """\
+            #!/usr/bin/env bash
+            set -eu
+
+            increment() {
+              counter_file="$FAKE_KUBECTL_STATE/$1.count"
+              COUNT=0
+              if [ -f "$counter_file" ]; then
+                COUNT=$(cat "$counter_file")
+              fi
+              COUNT=$((COUNT + 1))
+              printf '%s' "$COUNT" > "$counter_file"
+            }
+
+            case "${1:-}" in
+              rollout)
+                increment rollout
+                exit 0
+                ;;
+              get)
+                increment get
+                if [[ " $* " == *" -o name "* ]]; then
+                  echo "pod/awoooi-worker-ready"
+                else
+                  echo "true"
+                fi
+                exit 0
+                ;;
+              wait)
+                increment wait
+                exit 0
+                ;;
+              exec)
+                increment exec
+                if [ "$COUNT" -le "$FAKE_KUBECTL_EXEC_FAILURES" ]; then
+                  echo 'container not found ("worker")' >&2
+                  exit 1
+                fi
+                cat <<'JSON'
+            {"role_fingerprint":"worker-fingerprint","connection_limit":8,"required_connection_budget":5,"active_role_connection_count":1,"available_connection_headroom":7,"representative_connection_probe_count":1,"representative_preflight_passed":true,"table_privilege_gap_count":0,"sequence_privilege_gap_count":0,"error_type":null}
+            JSON
+                exit 0
+                ;;
+            esac
+            exit 2
+            """
+        )
+    )
+    fake_kubectl.chmod(0o755)
+    shell = "\n".join(
+        (
+            "set -e",
+            'KUBECTL="$FAKE_KUBECTL"',
+            "AWOOOI_DB_IDENTITY_RETRY_BACKOFF_SECONDS=0",
+            _read_workload_db_identity_shell(),
+            "if result=$(read_workload_db_identity " "awoooi-worker worker 1 5); then",
+            "  status=0",
+            "else",
+            "  status=$?",
+            "fi",
+            'printf \'status=%s\\n%s\\n\' "$status" "$result"',
+        )
+    )
+    env = {
+        **os.environ,
+        "FAKE_KUBECTL": str(fake_kubectl),
+        "FAKE_KUBECTL_STATE": str(state_dir),
+        "FAKE_KUBECTL_EXEC_FAILURES": str(exec_failures),
+    }
+    completed = subprocess.run(
+        ["bash", "-c", shell],
+        check=True,
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+    counts = {
+        name: int((state_dir / f"{name}.count").read_text())
+        for name in ("rollout", "get", "wait", "exec")
+    }
+    return completed, counts
+
+
 def _deployment_container(path: str, container_name: str) -> dict:
     documents = yaml.safe_load_all((REPO_ROOT / path).read_text())
     deployment = next(
@@ -46,9 +173,7 @@ def test_runtime_workloads_use_distinct_database_secret_projections() -> None:
     for workload_id, (path, container_name, expected_secret) in workload_specs.items():
         container = _deployment_container(path, container_name)
         secret_env_from = [
-            source
-            for source in container.get("envFrom", [])
-            if source.get("secretRef")
+            source for source in container.get("envFrom", []) if source.get("secretRef")
         ]
         assert secret_env_from == [], workload_id
 
@@ -123,7 +248,7 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
     assert "DROP ROLE" not in script.upper()
     assert "base64 -d" not in script
     assert "GITHUB_RUN_ID" not in script
-    assert 'check|apply|verify|disable' in script
+    assert "check|apply|verify|disable" in script
     assert '"secret_value_exposed": False' in script
     assert '"secret_value_logged": False' in script
     assert "secret_value_read" not in script
@@ -137,11 +262,165 @@ def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     assert "restore_workload_database_projection" in workflow
     assert "bash -s -- apply" in workflow
     assert "bash -s -- verify" in workflow
-    assert workflow.index("bash -s -- apply") < workflow.index(
-        "bash -s -- verify"
-    )
+    assert workflow.index("bash -s -- apply") < workflow.index("bash -s -- verify")
     assert '"spec": item["spec"]' in workflow
     assert '"data": item' not in workflow
     assert "get secret '${secret_name}' -n awoooi-prod -o jsonpath" not in workflow
     assert "get secret awoooi-api-db -n awoooi-prod -o jsonpath" not in workflow
     assert "HEAD^" not in workflow
+    assert 'while [ "$attempt" -le 3 ]' in workflow
+    assert 'rollout status "deployment/$workload"' in workflow
+    assert 'wait --for=condition=Ready "$ready_pod_ref"' in workflow
+    assert 'exec -i "$ready_pod_ref"' in workflow
+    assert "container_not_ready" in workflow
+    assert "WorkloadDbIdentityAttemptsExhausted" in workflow
+    assert "workload DB identity attempts exhausted before receipt write" in workflow
+    assert "API_PUBLIC_READINESS_ATTEMPTS" in workflow
+    assert "API_HTTP_CONCURRENCY_ATTEMPTS" in workflow
+    assert "API sequential readiness permanent HTTP status" in workflow
+    assert "events/dossier/coverage?project_id=awoooi&limit=1" in workflow
+    assert "platform/cicd/events?project_id=awoooi&limit=1" in workflow
+    assert workflow.index("API_HTTP_CONCURRENCY_VERIFIED=$(python3") < workflow.index(
+        "$KUBECTL create configmap awoooi-executor-boundary-verification"
+    )
+
+
+def test_cd_workload_db_identity_retries_transient_worker_rollout_race(
+    tmp_path: Path,
+) -> None:
+    completed, counts = _run_workload_db_identity_contract(
+        tmp_path,
+        exec_failures=1,
+    )
+
+    lines = completed.stdout.strip().splitlines()
+    assert lines[0] == "status=0"
+    payload = json.loads(lines[-1])
+    assert payload["role_fingerprint"] == "worker-fingerprint"
+    assert payload["available_connection_headroom"] == 7
+    assert payload["representative_connection_probe_count"] == 1
+    assert payload["representative_preflight_passed"] is True
+    assert counts == {"rollout": 2, "get": 4, "wait": 2, "exec": 2}
+    assert "reason=kubectl_exec_failed" in completed.stderr
+    assert "workload_db_identity_exec_retry=1/3" in completed.stderr
+    assert "workload_db_identity_blocked" not in completed.stderr
+
+
+def test_cd_workload_db_identity_blocks_after_three_worker_exec_failures(
+    tmp_path: Path,
+) -> None:
+    completed, counts = _run_workload_db_identity_contract(
+        tmp_path,
+        exec_failures=3,
+    )
+
+    lines = completed.stdout.strip().splitlines()
+    assert lines[0] == "status=1"
+    payload = json.loads(lines[-1])
+    assert payload["role_fingerprint"] == ""
+    assert payload["representative_preflight_passed"] is False
+    assert payload["error_type"] == "WorkloadDbIdentityAttemptsExhausted"
+    assert counts == {"rollout": 3, "get": 6, "wait": 3, "exec": 3}
+    assert "workload_db_identity_exec_retry=3/3" in completed.stderr
+    assert "workload_db_identity_exec_exhausted=3" in completed.stderr
+
+
+def test_cd_api_readiness_retries_http_503_then_runs_all_200_concurrency(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    responses = [503, 200, 200, 200, *([200] * 12)]
+    calls: list[str] = []
+
+    def fake_urlopen(url: str, *, timeout: int) -> _FakeHTTPResponse:
+        calls.append(f"{url}|{timeout}")
+        status = responses.pop(0)
+        if status == 503:
+            raise urllib.error.HTTPError(url, status, "unavailable", None, None)
+        return _FakeHTTPResponse(status)
+
+    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(time, "sleep", lambda _seconds: None)
+    monkeypatch.setenv("API_PUBLIC_READINESS_ATTEMPTS", "3")
+    monkeypatch.setenv("API_PUBLIC_READINESS_DELAY_SECONDS", "0")
+    monkeypatch.setenv("API_HTTP_CONCURRENCY_ATTEMPTS", "2")
+    monkeypatch.setenv("API_HTTP_CONCURRENCY_DELAY_SECONDS", "0")
+
+    exec(
+        compile(
+            _api_http_readiness_and_concurrency_python(),
+            "<cd-api-http-verifier>",
+            "exec",
+        ),
+        {},
+    )
+
+    output = capsys.readouterr()
+    assert output.out.strip() == "true"
+    assert "api_sequential_readiness=ready;attempt=2" in output.err
+    assert len(calls) == 16
+    assert responses == []
+
+
+def test_cd_api_readiness_exhaustion_stops_before_green_receipt_write(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    calls: list[str] = []
+
+    def always_unavailable(url: str, *, timeout: int) -> _FakeHTTPResponse:
+        calls.append(f"{url}|{timeout}")
+        raise urllib.error.HTTPError(url, 503, "unavailable", None, None)
+
+    monkeypatch.setattr(urllib.request, "urlopen", always_unavailable)
+    monkeypatch.setattr(time, "sleep", lambda _seconds: None)
+    monkeypatch.setenv("API_PUBLIC_READINESS_ATTEMPTS", "3")
+    monkeypatch.setenv("API_PUBLIC_READINESS_DELAY_SECONDS", "0")
+    namespace = {"receipt_write_reached": False}
+    verifier = _api_http_readiness_and_concurrency_python()
+
+    with pytest.raises(SystemExit, match="API sequential readiness failed"):
+        exec(
+            compile(
+                verifier + "\nreceipt_write_reached = True\n",
+                "<cd-api-http-verifier>",
+                "exec",
+            ),
+            namespace,
+        )
+
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert namespace["receipt_write_reached"] is False
+    assert len(calls) == 6
+
+
+def test_cd_api_readiness_permanent_http_error_fails_without_retry(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    calls: list[str] = []
+
+    def unauthorized(url: str, *, timeout: int) -> _FakeHTTPResponse:
+        calls.append(f"{url}|{timeout}")
+        raise urllib.error.HTTPError(url, 401, "unauthorized", None, None)
+
+    monkeypatch.setattr(urllib.request, "urlopen", unauthorized)
+    monkeypatch.setattr(time, "sleep", lambda _seconds: None)
+    namespace = {"receipt_write_reached": False}
+    verifier = _api_http_readiness_and_concurrency_python()
+
+    with pytest.raises(SystemExit, match="permanent HTTP status: 401"):
+        exec(
+            compile(
+                verifier + "\nreceipt_write_reached = True\n",
+                "<cd-api-http-verifier>",
+                "exec",
+            ),
+            namespace,
+        )
+
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert namespace["receipt_write_reached"] is False
+    assert len(calls) == 1