diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 729c0c53d..45ca4c838 100644 --- a/.gitea/workflows/cd.yaml +++ b/.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 diff --git a/apps/api/tests/test_workload_database_identity_projection.py b/apps/api/tests/test_workload_database_identity_projection.py index 1ffde1910..be6dff27c 100644 --- a/apps/api/tests/test_workload_database_identity_projection.py +++ b/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(), + "", + "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", + "", + "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", + "", + "exec", + ), + namespace, + ) + + output = capsys.readouterr() + assert output.out == "" + assert namespace["receipt_write_reached"] is False + assert len(calls) == 1