diff --git a/.github/workflows/deploy-pilot-no-traffic-protected.yml b/.github/workflows/deploy-pilot-no-traffic-protected.yml index c92e9b64..e516ccc6 100644 --- a/.github/workflows/deploy-pilot-no-traffic-protected.yml +++ b/.github/workflows/deploy-pilot-no-traffic-protected.yml @@ -82,6 +82,7 @@ jobs: PILOT_PROFILE: ${{ inputs.profile }} PILOT_SOURCE_COMMIT: ${{ inputs.source_commit }} PILOT_CLEANUP_RUN_ID: ${{ inputs.isolated_cleanup_run_id }} + PILOT_NETWORK_PREFLIGHT: ${{ inputs.isolated_network_preflight }} PILOT_JOB_NETWORK_PREFLIGHT: ${{ inputs.isolated_job_network_preflight }} PILOT_JOB_PHASE: ${{ inputs.isolated_job_phase }} PILOT_TEMPLATE_DIAGNOSTIC: ${{ inputs.isolated_template_diagnostic }} diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index b2c2cd5e..55c04f30 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -402,14 +402,24 @@ It changes no Scheduler, environment, Secret, IAM or runtime execution switch. The candidate is deployed with `--no-traffic --tag=cand- --min-instances=1` when the service name plus the candidate tag fits Cloud Run's -46-character combined limit. Longer names omit `--tag` entirely; promotion still -uses the exact qualified revision name, and cleanup skips tag removal. Short +46-character combined limit. When the preferred tag does not fit, the shared +policy allocates a shorter operation-bound tag if at least one character fits. +Only tag-impossible service names omit `--tag` entirely. After verifying the +unchanged no-traffic baseline, the exact-target traffic-approved lifecycle +assigns 99% to the previous revision +and 1% to the exact warm candidate, asserting the split readback before health +qualification. Promotion uses the exact qualified revision name. On failure +before promotion, cleanup recognizes only this exact split, restores the previous +revision to 100% and verifies it before resetting the minimum. Unexpected traffic +drift blocks cleanup mutations. Cleanup skips tag removal for untagged services. Short configured service names use the tagged path. This is a **revision-level** minimum, never service-level `--min`. Google Cloud's **Revisions and minimum instances** section says revision minimums start when referenced in a traffic split or assigned a tag; an untagged zero-percent revision alone does not establish startup. The untagged path retains -`--deploy-health-check` and all existing fresh health and baseline checks; missing -startup evidence stops promotion rather than routing traffic to warm the revision. +`--deploy-health-check` and strict fresh startup health checks after the approved +micro-split starts the revision. A Retired revision with no authentic instance +startup evidence cannot pass qualification. The tagged path retains zero-percent +serving traffic during qualification and verifies its baseline after health. One minimum instance is billed during this warm-up window, including idle time. Missing ContainerHealthy remains a hard denial, with bounded fresh readbacks. Raw condition readbacks separate absent startup evidence from an observed ContainerHealthy=False probe @@ -487,3 +497,28 @@ the candidate. Missing controls or invalid targets fail validation. Notification runtime-switch and initial-state recovery modes require explicit `dry_run=false`. Only a separately authorized execution may set `dry_run=false`; all existing main-ref, admission and approval gates still apply. + +### Shared revision-tag preflight + +PR #976 is source-only. Peer4 is excluded from its service scope. Merging this +change authorizes no deployment or redeployment; any later deployment requires +a separate owner card naming the exact service, revision and permitted actions. + +Operator entry points use `scripts/cloud_run_tag_policy.py`, a local Option B +implementation of `cloud-run-tags/v1`; QuantPlatformKit is the intended canonical +owner. The policy validates the exact resolved service name and the combined +46-character service/tag budget. It keeps a fitting preferred tag, otherwise +uses a deterministic operation-bound shortened tag. Occupied tags are refused, +including tags on zero-percent routes; a short name alone proves no ownership. +The sync plan reports `tag_preflight` without changing service identity. + +Only the configured candidate warmup lifecycle accepts an exact-service warmup +traffic approval for the temporary 99/1 split when no nonempty tag fits. TCP and +maintenance-cash probes remain strictly no-traffic and refuse such a target. +Image-only and notification-only entry points validate names without acquiring +traffic authority. Existing workflow concurrency, immutable image/identity +checks, baseline readbacks and cleanup remain required. Single-use local claims +are retained after interruption and success: reconcile the existing attempt +rather than deleting a claim or replaying an uncertain mutation. Local claims +do not provide a distributed lock; provider resource-version checks and the +protected workflow concurrency boundary remain required for cloud execution. diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index 9f9f52e7..8d3a8de5 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -60,6 +60,7 @@ def _should_add_local_src(candidate: Path) -> bool: from quant_platform_kit.common.runtime_heartbeat_policy import ( # noqa: E402 runtime_target_permits_standard_execution, ) +from scripts import cloud_run_tag_policy as tag_policy # noqa: E402 from scripts.reconciliation_recovery_state_ledger import ( # noqa: E402 apply_recovery_state_ledger_from_env, ) @@ -539,6 +540,9 @@ def _build_target_plan( if not service_name: raise ValueError("Each Cloud Run sync target requires service/service_name") + # Validate the exact configured name; budget classification is not authority + # to rename a target or start an untagged no-traffic candidate. + tag_preflight = tag_policy.preflight(service_name, "cand-" + "0" * 8) runtime_target_service = str(runtime_target.get("service_name") or "").strip() if runtime_target_service and runtime_target_service != service_name: raise ValueError( @@ -568,6 +572,7 @@ def _build_target_plan( permitted = enabled and runtime_target_permits_standard_execution(runtime_target) return { "service_name": service_name, + "tag_preflight": tag_preflight, "strategy_profile": raw_profile, "env": {"RUNTIME_TARGET_ENABLED": "true" if enabled else "false", "RUNTIME_TARGET_JSON": json.dumps(runtime_target, separators=(",", ":"), sort_keys=True)}, @@ -740,6 +745,7 @@ def _build_target_plan( return { "service_name": service_name, + "tag_preflight": tag_preflight, "strategy_profile": canonical_profile, "env": env_values, "scheduler": scheduler, diff --git a/scripts/cloud_run_candidate_warmup.py b/scripts/cloud_run_candidate_warmup.py index a1d0e80f..5468e899 100644 --- a/scripts/cloud_run_candidate_warmup.py +++ b/scripts/cloud_run_candidate_warmup.py @@ -18,6 +18,7 @@ from notifications.redaction import redact_notification_diagnostics from scripts import reconcile_cloud_runtime as reconcile +from scripts import cloud_run_tag_policy as tag_policy from scripts.runtime_target_continuity import verify_traffic_revision MINIMUM = 'autoscaling.knative.dev/minScale' @@ -66,26 +67,32 @@ def active(service: dict) -> list: class Release: def __init__(self, *, project: str, region: str, service: str, commit: str, digest: str, run_id: str, deploy_args: list[str], state_path: Path, - cloud: Callable[[list[str]], dict] = cloud, no_traffic_verify: Callable[[], None] | None = None): + cloud: Callable[[list[str]], dict] = cloud, no_traffic_verify: Callable[[], None] | None = None, + warmup_traffic_approved: bool = False, approved_service: str | None = None): self.project, self.region, self.service = project, region, service self.commit, self.digest, self.run_id = commit, digest, run_id self.deploy_args, self.state_path, self.cloud = deploy_args, state_path, cloud self.no_traffic_verify = no_traffic_verify - tag = 'cand-' + commit[:8] - # Cloud Run limits the combined tag and service name to 46 characters. - # Long exact targets use revision identity without a traffic tag. - self.tag = tag if len(service) + len(tag) <= 46 else None + self.warmup_traffic_approved = warmup_traffic_approved + self.approved_service = approved_service + self.operation_id = run_id + ':' + commit + ':' + digest + self.allocation = tag_policy.allocate_tag(service, 'cand-' + commit[:8], + existing_tags=[], operation_id=self.operation_id) + self.tag = self.allocation.tag self.common = [f'--project={project}', f'--region={region}', '--format=json'] self.warm = None self.reset = None self.expected = None self.before = None self.cleaning = False + self.micro_traffic = None def record(self, status: str) -> None: # No resource names, bindings, env or provider diagnostics in this result. self.state_path.write_text(json.dumps({'status': status, 'tag': self.tag, - 'reset_creates_revision': True}) + '\n') + 'reset_creates_revision': True, + 'contract_version': tag_policy.CONTRACT_VERSION, + 'operation_fingerprint': self.allocation.operation_fingerprint}) + '\n') def service_read(self) -> dict: return self.cloud(['gcloud', 'run', 'services', 'describe', self.service, *self.common]) @@ -128,7 +135,9 @@ def readback(self, stage: str) -> None: def stable(self) -> dict: current = self.service_read() - if self.expected is None or current != self.expected: + if (self.expected is None or current != self.expected + or (self.before is not None and current.get('metadata', {}).get('uid') + != self.before.get('metadata', {}).get('uid'))): raise RuntimeError('candidate service changed outside this attempt') return current @@ -151,7 +160,9 @@ def health(self, value: dict) -> None: reconcile._verify_deploy_candidate_with_wait( value, project=self.project, region=self.region, service_name=self.service, service=current, revision=value['metadata']['name'], commit=self.commit, - require_generation=True, wait_for_start=True) + require_generation=True, wait_for_start=True, + **({'allow_warmup_split': True} + if self.micro_traffic is not None and value['metadata']['name'] == self.warm else {})) self.stable() # A fresh read after the wait prevents a cached healthy value from # authorizing a later update. Existing continuity checks stay strict. @@ -161,6 +172,25 @@ def health(self, value: dict) -> None: verify_traffic_revision(fresh, expected_revision=value['metadata']['name'], expected_commit=self.commit, require_startup_health=True) + def start_untagged(self) -> None: + if tag_policy.lifecycle_for_impossible_tag(service_name=self.service, + approved_service=self.approved_service, + warmup_traffic_approved=self.warmup_traffic_approved) != 'MICRO_SPLIT': + raise RuntimeError('TAG_IMPOSSIBLE: exact warmup approval required') + self.stable() + previous = active(self.before)[0][0] + self.micro_traffic = sorted([(previous, 99), (self.warm, 1)]) + try: + self.cloud(['gcloud', 'run', 'services', 'update-traffic', self.service, *self.common, + '--to-revisions=' + previous + '=99,' + self.warm + '=1', '--quiet']) + finally: + # A timeout may follow an applied split. Read once, never replay; + # cleanup accepts only the exact original or intended distribution. + self.expected = self.service_read() + if (self.expected.get('metadata', {}).get('uid') != self.before['metadata']['uid'] + or active(self.expected) != self.micro_traffic): + raise RuntimeError('candidate micro traffic split readback mismatch') + def promote(self, value: dict) -> None: self.stable() # Baseline validation can outlast the earlier health observation. @@ -186,7 +216,18 @@ def restore(self) -> str | None: self.warm = warm['metadata']['name'] serving = active(current) == [(self.warm, 100)] if not serving and active(current) != active(self.before): - raise RuntimeError('candidate cleanup traffic identity changed') + if self.micro_traffic is None or active(current) != self.micro_traffic: + raise RuntimeError('candidate cleanup traffic identity changed') + self.stable() + previous = active(self.before)[0][0] + self.cloud(['gcloud', 'run', 'services', 'update-traffic', self.service, *self.common, + '--to-revisions=' + previous + '=100', '--quiet']) + self.expected = self.service_read() + if (self.expected.get('metadata', {}).get('uid') != current.get('metadata', {}).get('uid') + or active(self.expected) != active(self.before)): + raise RuntimeError('candidate cleanup prior traffic restore mismatch') + self.candidate(self.expected, '1') + current = self.expected tag_removed = True if self.tag: rows = current['status']['traffic'] @@ -243,6 +284,10 @@ def execute(self) -> str | None: raise ValueError('candidate deploy target binding mismatch') if '--no-traffic' not in self.deploy_args or '--deploy-health-check' not in self.deploy_args: raise ValueError('candidate deploy requires no-traffic and health check') + if self.tag is None and tag_policy.lifecycle_for_impossible_tag( + service_name=self.service, approved_service=self.approved_service, + warmup_traffic_approved=self.warmup_traffic_approved) != 'MICRO_SPLIT': + raise RuntimeError('TAG_IMPOSSIBLE: exact warmup approval required') # Exclusive create persists across crashes/restarts; no automatic replay. try: fd = os.open(self.state_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) @@ -259,7 +304,10 @@ def execute(self) -> str | None: previous = self.revision_read(initial_traffic[0][0]) if str(previous.get('metadata', {}).get('annotations', {}).get(MINIMUM, '0')) != '0': raise RuntimeError('candidate previous serving revision minimum is not zero') - if self.tag and any(row.get('tag') == self.tag for row in self.before['status']['traffic']): + allocation = tag_policy.allocate_tag(self.service, 'cand-' + self.commit[:8], + existing_tags=[row['tag'] for row in self.before['status']['traffic'] if row.get('tag')], + operation_id=self.operation_id) + if allocation.status == 'REFUSE' or allocation != self.allocation: raise RuntimeError('candidate tag collision; no mutation') if str(self.before.get('spec', {}).get('template', {}).get('metadata', {}).get('annotations', {}).get(MINIMUM, '0')) != '0': raise RuntimeError('candidate requires previous revision minimum zero') @@ -280,11 +328,17 @@ def execute(self) -> str | None: self.warm = warm['metadata']['name'] if active(self.expected) != active(self.before): raise RuntimeError('candidate deploy unexpectedly moved traffic') - self.readback('set') - self.health(warm) if self.no_traffic_verify is None: raise RuntimeError('candidate original no-traffic verification missing') - self.no_traffic_verify() + if self.tag is None: + # Verify the unchanged baseline before temporarily routing 1%. + self.no_traffic_verify() + self.start_untagged() + warm = self.candidate(self.expected, '1') + self.readback('set') + self.health(warm) + if self.tag is not None: + self.no_traffic_verify() self.stable() self.promote(warm) self.record('promoted') @@ -334,7 +388,8 @@ def main(argv=None) -> int: service=os.environ['INPUT_CONFIGURED_SERVICE'], commit=os.environ['SOURCE_COMMIT'], digest=args.image.split('@')[-1], run_id=os.environ['GITHUB_RUN_ID'], deploy_args=args.deploy_args[1:] if args.deploy_args[:1] == ['--'] else args.deploy_args, - state_path=args.state_path, no_traffic_verify=lambda: _verify(baseline_args)) + state_path=args.state_path, no_traffic_verify=lambda: _verify(baseline_args), + warmup_traffic_approved=True, approved_service=baseline_args.service) def interrupted(signum, frame): # A catchable abort enters finally. Once cleanup starts, finish its # bounded attempts instead of interrupting the restoration itself. diff --git a/scripts/cloud_run_tag_policy.py b/scripts/cloud_run_tag_policy.py new file mode 100644 index 00000000..4a4b49e8 --- /dev/null +++ b/scripts/cloud_run_tag_policy.py @@ -0,0 +1,91 @@ +"""Versioned Option B operator policy; intended canonical owner: QuantPlatformKit. + +Pure allocation is not traffic authority. Callers persist exclusive no-replay +state, re-read current routing, use provider CAS where available, and verify +owned tag removal plus restoration. Never infer ownership from a short tag. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +import hashlib +import json +import os +from pathlib import Path +import re + +CONTRACT_VERSION = "cloud-run-tags/v1" +TAGGED_URL_BUDGET = 46 +SERVICE = re.compile(r"[a-z](?:[a-z0-9-]{0,47}[a-z0-9])?") +TAG = re.compile(r"[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?") + + +@dataclass(frozen=True) +class Allocation: + status: str + tag: str | None = field(repr=False) + operation_fingerprint: str = field(repr=False) + + +def headroom(service_name: str) -> int: + if not isinstance(service_name, str) or not SERVICE.fullmatch(service_name): + raise ValueError("INVALID_SERVICE_NAME") + return TAGGED_URL_BUDGET - len(service_name) + + +def allocate_tag(service_name: str, preferred_tag: str, *, existing_tags, + operation_id: str) -> Allocation: + room = headroom(service_name) + if (not isinstance(preferred_tag, str) or not TAG.fullmatch(preferred_tag) + or not isinstance(operation_id, str) or not operation_id + or not isinstance(existing_tags, (list, tuple, set, frozenset)) + or any(not isinstance(tag, str) or not TAG.fullmatch(tag) for tag in existing_tags)): + raise ValueError("INVALID_TAG_ALLOCATION_INPUT") + fingerprint = hashlib.sha256((CONTRACT_VERSION + "\0" + service_name + "\0" + + preferred_tag + "\0" + operation_id).encode()).hexdigest() + if room <= 0: + return Allocation("TAG_IMPOSSIBLE", None, fingerprint) + # A leading letter is required even when the entire budget is one byte. + shortened = chr(ord('a') + int(fingerprint[:2], 16) % 26) + fingerprint[2:] + tag = preferred_tag if len(preferred_tag) <= room else shortened[:room] + if tag in existing_tags: + # No implicit same-owner reuse: persisted interrupted state needs + # reconciliation, and an occupied tag alone never proves ownership. + return Allocation("REFUSE", None, fingerprint) + return Allocation("ALLOCATED", tag, fingerprint) + + +def lifecycle_for_impossible_tag(*, service_name: str, approved_service: str | None = None, + warmup_traffic_approved: bool = False, + no_traffic: bool = False) -> str: + headroom(service_name) + if (headroom(service_name) <= 0 and warmup_traffic_approved is True + and no_traffic is False and approved_service == service_name): + return "MICRO_SPLIT" + return "STOP" + + +def preflight(service_name: str, preferred_tag: str | None = None) -> dict: + """Identity passthrough: classify budget without authorizing any lifecycle.""" + room = headroom(service_name) + if preferred_tag is None: + return {"contract_version": CONTRACT_VERSION, "headroom": room, + "tag_status": "TAG_IMPOSSIBLE" if room <= 0 else "TAG_AVAILABLE"} + if not isinstance(preferred_tag, str) or not TAG.fullmatch(preferred_tag): + raise ValueError("INVALID_TAG_ALLOCATION_INPUT") + status = ("TAG_IMPOSSIBLE" if room <= 0 else + "TAG_FITS" if len(preferred_tag) <= room else "TAG_SHORTEN_REQUIRED") + return {"contract_version": CONTRACT_VERSION, "headroom": room, + "tag_status": status} + + +def claim_operation(path: Path, operation_id: str) -> None: + """Single-use entry point claim; keep after success/failure for reconciliation.""" + try: + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + raise ValueError("operation restart requires reconciliation; no replay") from None + with os.fdopen(fd, "w") as stream: + json.dump({"contract_version": CONTRACT_VERSION, + "operation_fingerprint": hashlib.sha256(operation_id.encode()).hexdigest()}, stream) + stream.flush() + os.fsync(stream.fileno()) diff --git a/scripts/deploy_notification_only.py b/scripts/deploy_notification_only.py index ac84477b..5d16993f 100644 --- a/scripts/deploy_notification_only.py +++ b/scripts/deploy_notification_only.py @@ -19,6 +19,8 @@ import tarfile import time +from scripts import cloud_run_tag_policy as tag_policy + from scripts.verify_cloud_run_no_traffic_deploy import _active_traffic, _digest from scripts.verify_image_ues_release_binding import check, new_report, revision_from_lockfile @@ -245,6 +247,10 @@ def resolve_service(raw: str) -> str: raise ReleaseError("protected_inventory_unavailable") from None matches = [name for name in names if isinstance(name, str) and sha(name) == SERVICE_HASH] require(len(matches) == 1 and bool(re.fullmatch(r"[a-z][a-z0-9-]{0,62}", matches[0])), "fixed_target_not_uniquely_resolved") + try: + tag_policy.preflight(matches[0]) + except ValueError: + raise ReleaseError("tag_preflight_invalid") from None return matches[0] diff --git a/scripts/deploy_pilot_no_traffic.py b/scripts/deploy_pilot_no_traffic.py index 9ceb241b..5eaa93e9 100644 --- a/scripts/deploy_pilot_no_traffic.py +++ b/scripts/deploy_pilot_no_traffic.py @@ -17,6 +17,8 @@ from typing import Any, Mapping from urllib.parse import urlsplit +from scripts import cloud_run_tag_policy as tag_policy + PROFILE = "tqqq_growth_income" REPOSITORY = "QuantStrategyLab/InteractiveBrokersPlatform" IDENTITY_FIELDS = ("service_name", "strategy_profile", "platform_id", "account_scope", "deployment_selector", "account_selector") @@ -119,6 +121,13 @@ def context(env: Mapping[str, str]) -> Context: service = next((str(_target_field(entry, defaults, key)).strip() for key in ("service", "service_name", "cloud_run_service") if _target_field(entry, defaults, key)), target.get("service_name", "")) + try: + tag_policy.preflight(service) + if env.get("PILOT_NETWORK_PREFLIGHT") == "true": + if tag_policy.preflight(service, "tcp-" + "0" * 8)["tag_status"] == "TAG_IMPOSSIBLE": + raise ReleaseError("TAG_IMPOSSIBLE") + except ValueError: + raise ReleaseError("tag_preflight_invalid") from None accounts = target.get("account_selector") if (not NAME.fullmatch(service) or target.get("service_name") != service or target.get("platform_id") != "ibkr" diff --git a/scripts/observe_maintenance_cash.py b/scripts/observe_maintenance_cash.py index 715d6d16..d2caef4a 100644 --- a/scripts/observe_maintenance_cash.py +++ b/scripts/observe_maintenance_cash.py @@ -16,6 +16,7 @@ from urllib.parse import urlsplit from scripts import deploy_pilot_no_traffic as pilot +from scripts import cloud_run_tag_policy as tag_policy from scripts import diagnose_account_facts_identity as identity from scripts import publish_account_facts_from_report as publisher @@ -24,21 +25,15 @@ def digest(value): return hashlib.sha256(value.encode()).hexdigest() -def traffic_tag(run_id, service): - # Cloud Run limits tag + service name to 46 characters. The full run ID - # remains in the Scheduler name. A shortened tag is never overwritten: - # the existing-resource check rejects any occupied name before mutation. - number, encoded = int(run_id), "" - while number: - number, digit = divmod(number, 36) - encoded = "0123456789abcdefghijklmnopqrstuvwxyz"[digit] + encoded - available = 46 - len(service) - if available < 2: - raise pilot.ReleaseError("candidate_tag_length_unsupported") - tag = "c" + (encoded or "0")[-(available - 1):] - if not pilot.NAME.fullmatch(tag): - raise pilot.ReleaseError("candidate_tag_length_unsupported") - return tag +def traffic_tag(run_id, service, *, existing_tags=()): + try: + allocation = tag_policy.allocate_tag(service, "cash-" + digest(run_id)[:8], + existing_tags=existing_tags, operation_id="maintenance-cash:" + run_id) + except ValueError: + raise pilot.ReleaseError("tag_preflight_invalid") from None + if allocation.status != "ALLOCATED": + raise pilot.ReleaseError(allocation.status) + return allocation.tag def configuration(revision): @@ -131,7 +126,8 @@ def read(args, stage): raise pilot.ReleaseError("candidate_not_admitted") revision = read(["run", "revisions", "describe", candidate, *common], "candidate_read_failed") candidate = admit(expected, before, revision, source) - tag = traffic_tag(env["GITHUB_RUN_ID"], ctx.service) + tag = traffic_tag(env["GITHUB_RUN_ID"], ctx.service, + existing_tags=[row["tag"] for row in before["traffic"] if row["tag"]]) job = "ibkr-cash-read-" + env["GITHUB_RUN_ID"] service_url = before["service"]["status"]["url"] # Reuse the existing warm-up identity. No new IAM, account or caller. @@ -148,6 +144,16 @@ def read(args, stage): if (any(row.get("name", "").rsplit("/", 1)[-1] == job for row in jobs) or any(row["tag"] == tag for row in before["traffic"])): raise pilot.ReleaseError("observation_resource_already_exists") + try: + fd = os.open(receipt.with_suffix(".claim"), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + raise pilot.ReleaseError("observation_restart_requires_reconciliation") from None + with os.fdopen(fd, "w") as claim: + claim.write(json.dumps({"contract_version": tag_policy.CONTRACT_VERSION, + "operation_fingerprint": digest(ctx.service + ":" + candidate + ":" + env["GITHUB_RUN_ID"]), + "tag": tag})) + claim.flush() + os.fsync(claim.fileno()) started = now() tag_attempted = job_attempted = False result = dict(status="STOPPED", reason="observation_not_completed", source_commit=source, @@ -161,6 +167,13 @@ def save(): save() try: + fresh = snapshot(ctx, run) + if fresh != before: + raise pilot.ReleaseError("observation_baseline_changed") + # Allocation uses the immediately re-read routing, never a saved batch. + if traffic_tag(env["GITHUB_RUN_ID"], ctx.service, + existing_tags=[row["tag"] for row in fresh["traffic"] if row["tag"]]) != tag: + raise pilot.ReleaseError("observation_tag_changed") tag_attempted = True run(["gcloud", "run", "services", "update-traffic", ctx.service, *common, "--update-tags=" + tag + "=" + candidate, "--quiet", "--format=json"], stage="tag_update_unknown") diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 35c27223..7522116d 100755 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -530,7 +530,8 @@ def _read_deploy_cloud(args: Sequence[str], *, json_output: bool = True, def _verify_deploy_candidate_with_wait(candidate: Mapping[str, Any], *, project: str, region: str, service_name: str, service: Mapping[str, Any], revision: str, commit: str | None, - require_generation: bool = False, wait_for_start: bool = False) -> None: + require_generation: bool = False, wait_for_start: bool = False, + allow_warmup_split: bool = False) -> None: """Require stable fresh startup health while old traffic stays fixed. Observations are bounded to 300 seconds, not an atomic Cloud Run traffic CAS. @@ -575,6 +576,17 @@ def identity(value: Mapping[str, Any]) -> str: ) from None raise return + if allow_warmup_split: + split = service.get("status", {}).get("traffic") + if isinstance(split, list): + # Existing zero-percent tags do not change the serving distribution. + # The common validation below still checks every traffic entry. + split = [row for row in split if row.get("percent", 0) != 0] + if (not warmed or not wait_for_start or not isinstance(split, list) or len(split) != 2 + or sorted(row.get("percent", 0) for row in split) != [1, 99] + or sum(row.get("revisionName") == revision and row.get("percent") == 1 + for row in split) != 1): + raise ReconcileError("candidate warmup requires exact approved micro split") baseline = json.dumps({"spec": service.get("spec", {}).get("traffic"), "status": service.get("status", {}).get("traffic")}, sort_keys=True) traffic = service.get("status", {}).get("traffic") @@ -582,7 +594,8 @@ def identity(value: Mapping[str, Any]) -> str: or any(type(row.get("percent", 0)) is not int or row.get("percent", 0) < 0 or not row.get("revisionName") for row in traffic) or sum(row.get("percent", 0) for row in traffic) != 100 - or any(row.get("revisionName") == revision and row.get("percent", 0) != 0 for row in traffic)): + or any(row.get("revisionName") == revision + and row.get("percent", 0) != (1 if allow_warmup_split else 0) for row in traffic)): raise ReconcileError("candidate warmup requires known old traffic and zero candidate traffic") deadline = time.monotonic() + 300 diff --git a/scripts/run_pilot_cloud_tcp.py b/scripts/run_pilot_cloud_tcp.py index da475c19..deeada26 100644 --- a/scripts/run_pilot_cloud_tcp.py +++ b/scripts/run_pilot_cloud_tcp.py @@ -17,6 +17,7 @@ import urllib.request from scripts import deploy_pilot_no_traffic as pilot +from scripts import cloud_run_tag_policy as tag_policy from scripts import diagnose_pilot_serving_install as serving from scripts import diagnose_runner_gateway_access as gateway from scripts import diagnose_serving_image_identity as identity @@ -87,7 +88,16 @@ def prepared(ctx, before, group, run_id): # Keep the existing repository and never adopt a newer serving digest. base = base.rsplit("@", 1)[0] + "@" + BASE candidate = ctx.service + "-p" + hashlib.sha256(run_id.encode()).hexdigest()[:8] - tag = "tcp-" + hashlib.sha256(run_id.encode()).hexdigest()[:8] + try: + allocation = tag_policy.allocate_tag(ctx.service, + "tcp-" + hashlib.sha256(run_id.encode()).hexdigest()[:8], + existing_tags=[row["tag"] for row in before["traffic"] if row["tag"]], + operation_id="tcp:" + run_id + ":" + candidate) + except ValueError: + raise pilot.ReleaseError("tag_preflight_invalid") from None + if allocation.status != "ALLOCATED": + raise pilot.ReleaseError(allocation.status) + tag = allocation.tag if not pilot.NAME.fullmatch(candidate): raise pilot.ReleaseError("temporary_name_unavailable") if any(row["tag"] == tag for row in before["traffic"]): @@ -166,7 +176,7 @@ def probe_configuration(template): return config -def execute(ctx, run_id, *, run=pilot.command, snapshot=pilot.snapshot, api=request, pause=time.sleep, progress=None): +def execute(ctx, run_id, *, state_path=None, run=pilot.command, snapshot=pilot.snapshot, api=request, pause=time.sleep, progress=None): out = result() if progress is None else progress before = snapshot(ctx, run) pilot._no_manual_scaling(before["service"]) @@ -180,6 +190,12 @@ def adapt(args, reason): group = gateway.read_group(ctx, source["group_declarations"], adapt) desired, candidate, code_hash = prepared(ctx, before, group, run_id) probe = (candidate, desired["spec"]["traffic"][-1]["tag"]) + if state_path is None: + raise pilot.ReleaseError("operation_state_required") + try: + tag_policy.claim_operation(state_path, ctx.project + ":" + ctx.region + ":" + ctx.service + ":" + candidate + ":" + run_id) + except ValueError: + raise pilot.ReleaseError("operation_restart_requires_reconciliation") from None out.update(binding_verified=True, probe_code_sha256=code_hash) token = run(["gcloud", "auth", "print-access-token", "--quiet"], stage="original_identity_unavailable").strip() endpoint = f"https://{ctx.region}-run.googleapis.com/apis/serving.knative.dev/v1/namespaces/{ctx.project}" @@ -411,7 +427,7 @@ def main(argv=None): raise pilot.ReleaseError("cleanup_run_id_invalid") out = cleanup_existing(ctx, cleanup_id, progress=out) else: - out = execute(ctx, os.environ["GITHUB_RUN_ID"], progress=out) + out = execute(ctx, os.environ["GITHUB_RUN_ID"], state_path=args.receipt.with_suffix(".claim"), progress=out) except CloudError as error: out.update(stage=str(error), cloud_http_status=error.http_status) except pilot.ReleaseError as error: diff --git a/tests/test_candidate_no_tag_release.py b/tests/test_candidate_no_tag_release.py index dcc02780..5a064bdd 100644 --- a/tests/test_candidate_no_tag_release.py +++ b/tests/test_candidate_no_tag_release.py @@ -7,14 +7,15 @@ from scripts.cloud_run_candidate_warmup import Release from tests.test_candidate_tag_release import Cloud, COMMIT, DIGEST -SERVICE = 'unit-' + 'x' * 41 + '62' +SERVICE = 'interactive-brokers-quant-live-fake12345-service' def release(tmp_path, cloud): return Release(project='', region='unit-region', service=SERVICE, commit=COMMIT, digest=DIGEST, run_id='123', deploy_args=['--image=fake@' + DIGEST, '--no-traffic', '--deploy-health-check'], - state_path=tmp_path / 'state.json', cloud=cloud, no_traffic_verify=lambda: None) + state_path=tmp_path / 'state.json', cloud=cloud, no_traffic_verify=lambda: None, + warmup_traffic_approved=True, approved_service=SERVICE) def assert_no_tags(cloud): @@ -28,6 +29,10 @@ def test_long_service_qualifies_both_revisions_and_promotes_by_name(tmp_path): observed = [] def health(value, **kwargs): observed.append(value['metadata']['name']) + if value['metadata']['name'].endswith('-warm'): + assert kwargs['service']['status']['traffic'] == [ + {'revisionName': SERVICE + '-old', 'percent': 99}, + {'revisionName': SERVICE + '-warm', 'percent': 1}] with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait', side_effect=health): result = release(tmp_path, cloud).execute() assert len(SERVICE) == 48 @@ -72,7 +77,9 @@ def drift(*args, **kwargs): with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait', side_effect=drift): with pytest.raises(RuntimeError): release(tmp_path, cloud).execute() - assert not any('update-traffic' in command or 'update' in command for command in cloud.commands) + shifts = [command for command in cloud.commands if 'update-traffic' in command] + assert len(shifts) == 1 # Only the approved split, before the external change. + assert not any('update' in command for command in cloud.commands) assert_no_tags(cloud) @@ -88,10 +95,115 @@ def test_long_service_readback_keeps_shared_redaction(tmp_path, capsys): assert_no_tags(cloud) -@pytest.mark.parametrize('length,tagged', [(33, True), (34, False), (48, False)]) +@pytest.mark.parametrize('length,tagged', [(33, True), (34, True), (45, True), (48, False)]) def test_combined_tag_length_preserves_supported_short_services(tmp_path, length, tagged): obj = release(tmp_path, Cloud()) obj = Release(project=obj.project, region=obj.region, service='x' * length, commit=COMMIT, digest=DIGEST, run_id='123', deploy_args=obj.deploy_args, state_path=obj.state_path, cloud=obj.cloud) assert bool(obj.tag) is tagged + + +def test_split_failure_restores_prior_traffic_before_minimum_reset(tmp_path): + cloud = Cloud(service_name=SERVICE) + obj = release(tmp_path, cloud) + def baseline(): + assert cloud.service['status']['traffic'] == [{'revisionName': SERVICE + '-old', 'percent': 100}] + assert not any('update-traffic' in c for c in cloud.commands) + obj.no_traffic_verify = baseline + with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait', + side_effect=ValueError('startup failed')): + with pytest.raises(ValueError, match='startup failed'): + obj.execute() + split = next(i for i, c in enumerate(cloud.commands) if '--to-revisions=' + SERVICE + '-old=99,' + SERVICE + '-warm=1' in c) + rollback = next(i for i, c in enumerate(cloud.commands) if '--to-revisions=' + SERVICE + '-old=100' in c) + reset = next(i for i, c in enumerate(cloud.commands) if '--min-instances=0' in c) + assert split < rollback < reset + + +@pytest.mark.parametrize('mismatch', ['split', 'rollback']) +def test_micro_split_readback_mismatch_fails_closed(tmp_path, mismatch): + cloud = Cloud(service_name=SERVICE) + def driver(command): + result = cloud(command) + if 'update-traffic' in command and any(v.startswith('--to-revisions=') for v in command): + is_split = any('=99,' in v for v in command) + if is_split == (mismatch == 'split'): + cloud.service['status']['traffic'] = [ + {'revisionName': SERVICE + '-old', 'percent': 98}, + {'revisionName': SERVICE + '-warm', 'percent': 2}] + return result + obj = release(tmp_path, driver) + with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait', + side_effect=ValueError('startup failed')) as health: + with pytest.raises(RuntimeError, match='cleanup'): + obj.execute() + if mismatch == 'split': + health.assert_not_called() + assert not any('--min-instances=0' in c for c in cloud.commands) + assert not any('--to-revisions=' + SERVICE + '-warm=100' in c for c in cloud.commands) + + +def test_untagged_retired_candidate_starts_only_after_split(tmp_path): + cloud = Cloud(service_name=SERVICE) + def driver(command, **kwargs): + result = cloud(command) + if 'deploy' in command: + cold = cloud.revisions[SERVICE + '-warm'] + cold['status']['conditions'] = [ + {'type': 'Ready', 'status': 'True', 'reason': 'Retired'}, + {'type': 'Active', 'status': 'False', 'reason': 'Retired'}, + {'type': 'ContainerReady', 'status': 'True'}, + {'type': 'MinInstancesProvisioned', 'status': 'False', 'reason': 'NotProvisioned'}] + if any('=99,' in v for v in command): + cloud.revisions[SERVICE + '-warm']['status']['conditions'] = [ + {'type': kind, 'status': 'True'} for kind in + ('Ready', 'Active', 'ContainerReady', 'MinInstancesProvisioned', 'ContainerHealthy')] + return result + # Use the real strict waiter and health verifier, with only external + # continuity reads stubbed. Retired + NotProvisioned cannot pass it. + with mock.patch('scripts.runtime_target_continuity.verify_target_continuity'), \ + mock.patch('scripts.reconcile_cloud_runtime._read_deploy_cloud', side_effect=driver), \ + mock.patch('scripts.reconcile_cloud_runtime.time.sleep'): + assert release(tmp_path, driver).execute() == SERVICE + '-reset' + + +@pytest.mark.parametrize('failure', ['split_timeout', 'ownership_loss']) +def test_ambiguous_split_and_ownership_loss_do_not_replay(tmp_path, failure): + cloud = Cloud(service_name=SERVICE) + def driver(command): + result = cloud(command) + if any('=99,' in v for v in command): + if failure == 'ownership_loss': + cloud.service['metadata']['uid'] = 'other-owner' + raise RuntimeError('ambiguous split') + return result + with pytest.raises(RuntimeError): + release(tmp_path, driver).execute() + shifts = [c for c in cloud.commands if 'update-traffic' in c] + assert len(shifts) == (2 if failure == 'split_timeout' else 1) + if failure == 'split_timeout': + assert cloud.service['status']['traffic'] == [{'revisionName': SERVICE + '-old', 'percent': 100}] + assert any('--min-instances=0' in c for c in cloud.commands) + else: + assert not any('--min-instances=0' in c for c in cloud.commands) + + +@pytest.mark.parametrize('length', [46, 48]) +def test_micro_split_preserves_existing_zero_percent_tags(tmp_path, length): + service_name = 'unit-' + 'x' * (length - 5) + cloud = Cloud(service_name=service_name) + old_tag = {'revisionName': service_name + '-old', 'percent': 0, 'tag': 't'} + cloud.service['status']['traffic'].append(old_tag) + obj = Release(project='', region='unit-region', service=service_name, + commit=COMMIT, digest=DIGEST, run_id='123', + deploy_args=['--image=fake@' + DIGEST, '--no-traffic', '--deploy-health-check'], + state_path=tmp_path / 'state.json', cloud=cloud, no_traffic_verify=lambda: None, + warmup_traffic_approved=True, approved_service=service_name) + assert obj.tag is None + with mock.patch('scripts.runtime_target_continuity.verify_target_continuity'), \ + mock.patch('scripts.reconcile_cloud_runtime._read_deploy_cloud', side_effect=lambda c, **kw: cloud(c)), \ + mock.patch('scripts.reconcile_cloud_runtime.time.sleep'): + assert obj.execute() == service_name + '-reset' + assert old_tag in cloud.service['status']['traffic'] + assert_no_tags(cloud) diff --git a/tests/test_candidate_tag_release.py b/tests/test_candidate_tag_release.py index 53b06a71..28832cbf 100644 --- a/tests/test_candidate_tag_release.py +++ b/tests/test_candidate_tag_release.py @@ -76,9 +76,11 @@ def __call__(self, command): if not self.service['status']['traffic']: self.service['status']['traffic'] = [{'revisionName': self.service_name + '-warm', 'percent': 100}] else: - target = next(v.split('=', 1)[1].rsplit('=', 1)[0] for v in command if v.startswith('--to-revisions=')) + targets = next(v.split('=', 1)[1] for v in command if v.startswith('--to-revisions=')) tags = [dict(r, percent=0) for r in self.service['status']['traffic'] if r.get('tag')] - self.service['status']['traffic'] = [{'revisionName': target, 'percent': 100}] + tags + self.service['status']['traffic'] = [ + {'revisionName': name, 'percent': int(percent)} + for name, percent in (row.rsplit('=', 1) for row in targets.split(','))] + tags self.service['metadata']['resourceVersion'] = str(len(self.commands)) else: pytest.fail('unexpected command') diff --git a/tests/test_cr_tag_adoption.py b/tests/test_cr_tag_adoption.py new file mode 100644 index 00000000..7028598f --- /dev/null +++ b/tests/test_cr_tag_adoption.py @@ -0,0 +1,182 @@ +"""Shared policy adoption regressions; fake targets and offline drivers only.""" +from dataclasses import replace +import importlib +from unittest import mock + +import pytest + +from scripts import cloud_run_candidate_warmup as warmup +from scripts import deploy_pilot_no_traffic as pilot +from scripts import observe_maintenance_cash as cash +from scripts import run_pilot_cloud_tcp as tcp +from tests.test_candidate_tag_release import Cloud, COMMIT, DIGEST +from tests.test_pilot_cloud_tcp import lifecycle # noqa: F401 + + +def policy(): + return importlib.import_module('scripts.cloud_run_tag_policy') + + +def release(tmp_path, service, **kwargs): + cloud = Cloud(service_name=service) + obj = warmup.Release(project='fake-project', region='fake-region', service=service, + commit=COMMIT, digest=DIGEST, run_id='123', state_path=tmp_path / 'state.json', + deploy_args=['--no-traffic', '--deploy-health-check'], cloud=cloud, + no_traffic_verify=lambda: None, **kwargs) + return obj, cloud + + +@pytest.mark.parametrize('length', [34, 38, 44, 45]) +def test_warmup_shortens_instead_of_using_micro_traffic(tmp_path, length): + obj, cloud = release(tmp_path, 's' * length) + assert obj.tag and len(obj.tag) + length <= 46 + with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait'): + obj.execute() + assert not any('=99,' in arg for command in cloud.commands for arg in command) + + +def test_impossible_warmup_without_exact_approval_refuses_before_mutation(tmp_path): + obj, cloud = release(tmp_path, 's' * 48) + with pytest.raises((ValueError, RuntimeError), match='approv|TAG_IMPOSSIBLE'): + obj.execute() + assert not any('deploy' in command or 'update-traffic' in command for command in cloud.commands) + + +def test_tcp_shortens_and_refuses_impossible_without_micro_traffic(lifecycle): # noqa: F811 + for length in (38, 45): + ctx = replace(lifecycle.ctx, service='s' * length) + desired, _, _ = tcp.prepared(ctx, lifecycle.before, lifecycle.group, '123') + tag = desired['spec']['traffic'][-1]['tag'] + assert len(tag) + length <= 46 + assert desired['spec']['traffic'][-1]['percent'] == 0 + ctx = replace(lifecycle.ctx, service='s' * 48) + with pytest.raises(pilot.ReleaseError, match='TAG_IMPOSSIBLE'): + tcp.prepared(ctx, lifecycle.before, lifecycle.group, '123') + + +def test_cash_supports_one_character_budget_with_shared_policy(): + assert len(cash.traffic_tag('123', 's' * 45)) == 1 + with pytest.raises(pilot.ReleaseError, match='TAG_IMPOSSIBLE'): + cash.traffic_tag('123', 's' * 48) + + +def test_refusal_diagnostics_do_not_expose_supplied_identity(): + private = 'invalid-UPPERCASE-private-name' + with pytest.raises(ValueError) as error: + policy().headroom(private) + assert private not in str(error.value) + + +def test_foreign_tag_and_unknown_snapshot_refuse(): + module = policy() + tag = module.allocate_tag('s' * 45, 'cand-aaaaaaaa', existing_tags=[], operation_id='fake-op') + assert tag.status == 'ALLOCATED' + conflict = module.allocate_tag('s' * 45, 'cand-aaaaaaaa', existing_tags=[tag.tag], operation_id='fake-op') + assert conflict.status == 'REFUSE' + with pytest.raises(ValueError): + module.allocate_tag('fake-service', 'cand-aaaaaaaa', existing_tags=None, operation_id='fake-op') + + +def test_warmup_fresh_foreign_tag_blocks_mutation(tmp_path): + obj, cloud = release(tmp_path, 's' * 45) + assert obj.tag + original = obj.cloud + reads = 0 + def driver(command): + nonlocal reads + if 'services' in command and 'describe' in command: + reads += 1 + if reads == 2: + cloud.service['status']['traffic'].append( + {'revisionName': cloud.service_name + '-old', 'percent': 0, 'tag': obj.tag}) + return original(command) + obj.cloud = driver + with pytest.raises(RuntimeError): + obj.execute() + assert not any('deploy' in command for command in cloud.commands) + + +def test_shortened_warmup_restart_does_not_reallocate_or_replay(tmp_path): + obj, cloud = release(tmp_path, 's' * 45) + assert obj.tag + with mock.patch('scripts.reconcile_cloud_runtime._verify_deploy_candidate_with_wait'): + obj.execute() + count = len(cloud.commands) + with pytest.raises(RuntimeError, match='restart'): + obj.execute() + assert len(cloud.commands) == count + + +def test_no_traffic_policy_never_acquires_traffic_authority(): + module = policy() + assert module.lifecycle_for_impossible_tag(service_name='s' * 48, + approved_service='s' * 48, warmup_traffic_approved=True, no_traffic=True) == 'STOP' + assert module.lifecycle_for_impossible_tag(service_name='s' * 48, + approved_service='other-service', warmup_traffic_approved=True) == 'STOP' + + +@pytest.mark.parametrize('service,valid', [('fake-service', True), ('s' * 48, True), + ('Invalid-name', False), ('bad-name-', False), ('s' * 50, False)]) +def test_planner_validates_exact_name_without_renaming(service, valid): + from scripts import build_cloud_run_env_sync_plan as planner + runtime = {'service_name': service, 'strategy_profile': 'fake-profile'} + with mock.patch.object(planner, '_resolve_runtime_target', return_value=runtime.copy()): + arguments = dict(target={'service': service}, defaults={}, + env={'INPUT_RUNTIME_SWITCH_ONLY': 'true', 'RUNTIME_TARGET_ENABLED': 'false'}, + status_rows={}, per_service_mode=False, platform_config={}) + if not valid: + with pytest.raises(ValueError, match='INVALID_SERVICE_NAME'): + planner._build_target_plan(**arguments) + else: + plan = planner._build_target_plan(**arguments) + assert plan['service_name'] == service + assert plan['tag_preflight']['headroom'] == 46 - len(service) + if len(service) >= 46: + assert plan['tag_preflight']['tag_status'] == 'TAG_IMPOSSIBLE' + + +def test_tcp_restart_claim_blocks_second_mutation(lifecycle): # noqa: F811 + lifecycle.execute() + count = len(lifecycle.calls) + with pytest.raises(pilot.ReleaseError, match='restart'): + lifecycle.execute() + assert len(lifecycle.calls) == count + + +def test_exclusive_claim_is_shared_across_processes(tmp_path): + import concurrent.futures + import multiprocessing + path = tmp_path / 'shared.claim' + # fork workers import the same real policy and race O_EXCL on one inode. + with concurrent.futures.ProcessPoolExecutor(max_workers=2, + mp_context=multiprocessing.get_context('fork')) as executor: + outcomes = list(executor.map(_try_claim, [str(path), str(path)])) + assert sorted(outcomes) == ['claimed', 'refused'] + + +def _try_claim(path): + from pathlib import Path + try: + policy().claim_operation(Path(path), 'fake-operation') + return 'claimed' + except ValueError: + return 'refused' + + +def test_cash_fresh_foreign_tag_blocks_mutation(tmp_path): + from tests.test_observe_maintenance_cash import Cloud as CashCloud, inputs, NOW + from tests.test_pilot_no_traffic_release import environment + state, revision, expected, _ = inputs() + driver = CashCloud(state, revision) + reads = 0 + def snapshot(*args): + nonlocal reads + reads += 1 + result = driver.snapshot(*args) + if reads == 2: + result['traffic'].append({'revision': 'fake-foreign', 'percent': 0, 'tag': 'foreign'}) + return result + result = cash.observe(expected, COMMIT, environment(), tmp_path / 'safe.json', + run=driver.run, snapshot=snapshot, now=lambda: NOW) + assert result['status'] == 'STOPPED' and result['reason'] == 'observation_baseline_changed' + assert not any(args[1:4] == ['run', 'services', 'update-traffic'] for args in driver.calls) diff --git a/tests/test_observe_maintenance_cash.py b/tests/test_observe_maintenance_cash.py index ab0fbb3b..7c124ee8 100644 --- a/tests/test_observe_maintenance_cash.py +++ b/tests/test_observe_maintenance_cash.py @@ -64,9 +64,9 @@ def run(self, args, **_): if args[1:4] == ["run", "services", "update-traffic"]: rows = self.state["service"]["status"]["traffic"] if any(arg.startswith("--update-tags=") for arg in args): - rows.append(dict(revisionName=CANDIDATE, percent=0, tag="c3f", url="https://c3f---synthetic.run.app")) + rows.append(dict(revisionName=CANDIDATE, percent=0, tag="cash-a665a459", url="https://cash-a665a459---synthetic.run.app")) else: - rows[:] = [row for row in rows if row.get("tag") != "c3f"] + rows[:] = [row for row in rows if row.get("tag") != "cash-a665a459"] return "{}" if args[1:4] == ["scheduler", "jobs", "list"]: return json.dumps(self.jobs) @@ -75,7 +75,7 @@ def run(self, args, **_): if args[1:4] == ["scheduler", "jobs", "create"]: self.jobs.append(dict(name="ibkr-cash-read-123", state="ENABLED", schedule="03 01 30 09 *", timeZone="Etc/UTC", attemptDeadline="300s", retryConfig={}, httpTarget=dict( - uri="https://c3f---synthetic.run.app/probe", httpMethod="POST", headers={"X-QSL-Maintenance-Cash": "1"}, + uri="https://cash-a665a459---synthetic.run.app/probe", httpMethod="POST", headers={"X-QSL-Maintenance-Cash": "1"}, oidcToken=dict(serviceAccountEmail="synthetic-caller", audience="https://synthetic.run.app")))) if self.failure == "create_unknown": raise pilot.ReleaseError("scheduler_create_unknown") @@ -174,12 +174,13 @@ def test_zero_traffic_retired_candidate_does_not_replace_service_ready_identity( def test_realistic_run_identifier_fits_combined_tag_limit_and_shorter_budget(): run_id = "40218930896" tag = observer.traffic_tag(run_id, "s" * 38) - assert len(tag) == 8 and int(tag[1:], 36) == int(run_id) + assert len(tag) == 8 assert len(tag) + 38 == 46 assert len(observer.traffic_tag(run_id, "s" * 39)) == 7 assert len(observer.traffic_tag(run_id, "s" * 44)) == 2 - with pytest.raises(pilot.ReleaseError, match="candidate_tag_length_unsupported"): - observer.traffic_tag(run_id, "s" * 45) + assert len(observer.traffic_tag(run_id, "s" * 45)) == 1 + with pytest.raises(pilot.ReleaseError, match="TAG_IMPOSSIBLE"): + observer.traffic_tag(run_id, "s" * 46) assert tag != observer.traffic_tag(str(int(run_id) + 1), "s" * 38) @@ -197,11 +198,11 @@ def test_candidate_report_cannot_use_live_report_namespace(change): def test_occupied_short_tag_is_rejected_without_replacing_it(tmp_path): state, revision, expected, _ = inputs() - state["traffic"].append(dict(revision=CANDIDATE, percent=0, tag="c3f", url="https://c3f---synthetic.run.app")) + state["traffic"].append(dict(revision=CANDIDATE, percent=0, tag="cash-a665a459", url="https://cash-a665a459---synthetic.run.app")) cloud = Cloud(state, revision) - state["service"]["status"]["traffic"].append(dict(revisionName=CANDIDATE, percent=0, tag="c3f", url="https://c3f---synthetic.run.app")) + state["service"]["status"]["traffic"].append(dict(revisionName=CANDIDATE, percent=0, tag="cash-a665a459", url="https://cash-a665a459---synthetic.run.app")) cloud.state = state - with pytest.raises(pilot.ReleaseError, match="observation_resource_already_exists"): + with pytest.raises(pilot.ReleaseError, match="REFUSE"): observer.observe(expected, COMMIT, {**environment(), "CLOUD_RUN_REGION": "us-central1"}, tmp_path / "safe.json", run=cloud.run, snapshot=cloud.snapshot, now=lambda: NOW) assert not any(args[1:4] == ["run", "services", "update-traffic"] for args in cloud.calls) diff --git a/tests/test_pilot_cloud_tcp.py b/tests/test_pilot_cloud_tcp.py index 606c54db..f88398c3 100644 --- a/tests/test_pilot_cloud_tcp.py +++ b/tests/test_pilot_cloud_tcp.py @@ -17,7 +17,7 @@ @pytest.fixture -def lifecycle(monkeypatch): +def lifecycle(monkeypatch, tmp_path): ctx = pilot.context(environment()) before = baseline() before["service"]["spec"]["traffic"] = [{"revisionName": SERVING, "percent": 100}] @@ -96,7 +96,7 @@ def api(url, *, method, token, body=None, stage=None): assert method == "DELETE" return {} - state.execute = lambda: cloud.execute(ctx, "123", run=run, snapshot=snapshot, api=api, + state.execute = lambda: cloud.execute(ctx, "123", state_path=tmp_path / "tcp.claim", run=run, snapshot=snapshot, api=api, pause=lambda _: None, progress=state.progress) return state @@ -305,7 +305,7 @@ def test_only_exact_owned_zero_percent_route_can_be_ignored_during_probe(lifecyc def test_existing_tag_is_never_reassigned_to_the_probe(lifecycle): body, _, _ = cloud.prepared(lifecycle.ctx, lifecycle.before, lifecycle.group, "123") lifecycle.before["traffic"][0]["tag"] = body["spec"]["traffic"][-1]["tag"] - with pytest.raises(pilot.ReleaseError, match="temporary_tag_in_use"): + with pytest.raises(pilot.ReleaseError, match="REFUSE"): cloud.prepared(lifecycle.ctx, lifecycle.before, lifecycle.group, "123")