fix(cd): tolerate bounded workload rollout races

apps/api/tests/test_workload_database_identity_projection.py

@@ -43,6 +43,7 @@ def _run_workload_db_identity_contract(
     tmp_path: Path,
     *,
     exec_failures: int,
+    retry_attempts: int = 3,
 ) -> tuple[subprocess.CompletedProcess[str], dict[str, int]]:
     state_dir = tmp_path / "state"
     state_dir.mkdir()
@@ -103,6 +104,7 @@ def _run_workload_db_identity_contract(
             "set -e",
             'KUBECTL="$FAKE_KUBECTL"',
             "AWOOOI_DB_IDENTITY_RETRY_BACKOFF_SECONDS=0",
+            f"AWOOOI_DB_IDENTITY_RETRY_ATTEMPTS={retry_attempts}",
             _read_workload_db_identity_shell(),
             "if result=$(read_workload_db_identity " "awoooi-worker worker 1 5); then",
             "  status=0",
@@ -271,7 +273,9 @@ def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     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 'AWOOOI_DB_IDENTITY_RETRY_ATTEMPTS:-6' in workflow
+    assert 'while [ "$attempt" -le "$retry_attempts" ]' in workflow
+    assert 'workload_db_identity_exec_retry=$attempt/$retry_attempts' 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
@@ -358,6 +362,25 @@ def test_cd_workload_db_identity_blocks_after_three_worker_exec_failures(
     assert "workload_db_identity_exec_exhausted=3" in completed.stderr
 
 
+def test_cd_workload_db_identity_survives_extended_rollout_race(
+    tmp_path: Path,
+) -> None:
+    completed, counts = _run_workload_db_identity_contract(
+        tmp_path,
+        exec_failures=3,
+        retry_attempts=6,
+    )
+
+    lines = completed.stdout.strip().splitlines()
+    assert lines[0] == "status=0"
+    payload = json.loads(lines[-1])
+    assert payload["role_fingerprint"] == "worker-fingerprint"
+    assert payload["representative_preflight_passed"] is True
+    assert counts == {"rollout": 4, "get": 8, "wait": 4, "exec": 4}
+    assert "workload_db_identity_exec_retry=3/6" in completed.stderr
+    assert "workload_db_identity_exec_exhausted" not in completed.stderr
+
+
 def test_cd_api_readiness_retries_http_503_then_runs_all_200_concurrency(
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],