From 3ade6491409f6e617d12af49abba40848e7861f1 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 5 Nov 2026 21:41:29 +0800 Subject: [PATCH 1/4] fix: Start untagged warm candidates with an approved micro split --- docs/ibkr-runtime-rollout.md | 15 ++-- scripts/cloud_run_candidate_warmup.py | 51 ++++++++++++-- scripts/reconcile_cloud_runtime.py | 13 +++- tests/test_candidate_no_tag_release.py | 95 +++++++++++++++++++++++++- tests/test_candidate_tag_release.py | 6 +- 5 files changed, 163 insertions(+), 17 deletions(-) diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index b2c2cd5e..ac2183f2 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -402,14 +402,21 @@ 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. Longer names omit `--tag` entirely. After verifying the unchanged no-traffic +baseline, the 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 diff --git a/scripts/cloud_run_candidate_warmup.py b/scripts/cloud_run_candidate_warmup.py index a1d0e80f..ab8babdd 100644 --- a/scripts/cloud_run_candidate_warmup.py +++ b/scripts/cloud_run_candidate_warmup.py @@ -73,7 +73,7 @@ def __init__(self, *, project: str, region: str, service: str, commit: str, dige 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. + # Untaggable targets need an approved temporary traffic split to start. self.tag = tag if len(service) + len(tag) <= 46 else None self.common = [f'--project={project}', f'--region={region}', '--format=json'] self.warm = None @@ -81,6 +81,7 @@ def __init__(self, *, project: str, region: str, service: str, commit: str, dige 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. @@ -128,7 +129,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 +154,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 +166,21 @@ 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: + 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 +206,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'] @@ -280,11 +311,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') diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 35c27223..e3abfe60 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,13 @@ def identity(value: Mapping[str, Any]) -> str: ) from None raise return + if allow_warmup_split: + split = service.get("status", {}).get("traffic") + 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 +590,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/tests/test_candidate_no_tag_release.py b/tests/test_candidate_no_tag_release.py index dcc02780..47b05961 100644 --- a/tests/test_candidate_no_tag_release.py +++ b/tests/test_candidate_no_tag_release.py @@ -7,7 +7,7 @@ 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): @@ -28,6 +28,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 +76,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) @@ -95,3 +101,88 @@ def test_combined_tag_length_preserves_supported_short_services(tmp_path, 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) 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') From feee4274708af38553c8a5fb5ce06f5fa3d6d4c8 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 5 Nov 2026 21:52:59 +0800 Subject: [PATCH 2/4] fix: Preserve zero-percent tags during warmup qualification --- scripts/reconcile_cloud_runtime.py | 4 ++++ tests/test_candidate_no_tag_release.py | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index e3abfe60..7522116d 100755 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -578,6 +578,10 @@ def identity(value: Mapping[str, Any]) -> str: 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 diff --git a/tests/test_candidate_no_tag_release.py b/tests/test_candidate_no_tag_release.py index 47b05961..46d9ade9 100644 --- a/tests/test_candidate_no_tag_release.py +++ b/tests/test_candidate_no_tag_release.py @@ -186,3 +186,22 @@ def driver(command): 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', [34, 45]) +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) + 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) From e54b83798ea8c16d1e35df8e604dcb126cd8018e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 5 Nov 2026 23:16:59 +0800 Subject: [PATCH 3/4] fix(ops): adopt shared Cloud Run tag policy on warmup callers --- .../deploy-pilot-no-traffic-protected.yml | 1 + docs/ibkr-runtime-rollout.md | 28 ++- scripts/build_cloud_run_env_sync_plan.py | 6 + scripts/cloud_run_candidate_warmup.py | 34 +++- scripts/cloud_run_tag_policy.py | 91 +++++++++ scripts/deploy_notification_only.py | 6 + scripts/deploy_pilot_no_traffic.py | 9 + scripts/observe_maintenance_cash.py | 45 +++-- scripts/run_pilot_cloud_tcp.py | 22 ++- tests/test_candidate_no_tag_release.py | 10 +- tests/test_cr_tag_adoption.py | 182 ++++++++++++++++++ tests/test_observe_maintenance_cash.py | 19 +- tests/test_pilot_cloud_tcp.py | 6 +- 13 files changed, 414 insertions(+), 45 deletions(-) create mode 100644 scripts/cloud_run_tag_policy.py create mode 100644 tests/test_cr_tag_adoption.py 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 ac2183f2..aaf76eee 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -402,8 +402,11 @@ 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. After verifying the unchanged no-traffic -baseline, the traffic-approved lifecycle assigns 99% to the previous revision +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 @@ -494,3 +497,24 @@ 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 + +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 ab8babdd..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,15 +67,18 @@ 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. - # Untaggable targets need an approved temporary traffic split to start. - 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 @@ -86,7 +90,9 @@ def __init__(self, *, project: str, region: str, service: str, commit: str, dige 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]) @@ -167,6 +173,10 @@ def health(self, value: dict) -> None: 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)]) @@ -274,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) @@ -290,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') @@ -371,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/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 46d9ade9..5a064bdd 100644 --- a/tests/test_candidate_no_tag_release.py +++ b/tests/test_candidate_no_tag_release.py @@ -14,7 +14,8 @@ 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): @@ -94,7 +95,7 @@ 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, @@ -188,7 +189,7 @@ def driver(command): assert not any('--min-instances=0' in c for c in cloud.commands) -@pytest.mark.parametrize('length', [34, 45]) +@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) @@ -197,7 +198,8 @@ def test_micro_split_preserves_existing_zero_percent_tags(tmp_path, length): 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) + 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)), \ 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") From 1aa0efee3711797364ca71b9ad7ebfe999865373 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 04:04:05 +0800 Subject: [PATCH 4/4] docs: Exclude Peer4 from warmup service scope Co-Authored-By: Codex --- docs/ibkr-runtime-rollout.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index aaf76eee..55c04f30 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -500,6 +500,10 @@ 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