diff --git a/docs/runtime-target-manifest.md b/docs/runtime-target-manifest.md index 4b537cc6..0512d5eb 100644 --- a/docs/runtime-target-manifest.md +++ b/docs/runtime-target-manifest.md @@ -174,6 +174,40 @@ stop-result tools require `GCP_PROJECT_ID` in their process environment. `STRATEGY_PLUGIN_ALLOWED_SIGNAL_PREFIXES` supplies any additional approved plugin-signal prefixes; no project-specific bucket is a built-in fallback. + +## Cloud Run tag preflight + +The sync-plan builder validates the exact configured service name without trimming +or renaming it. Each target includes `tag_preflight` with the versioned shared +`cloud-run-tags/v1` contract, remaining tagged-URL headroom (`46 - len(service)`), +`AVAILABLE` or `TAG_IMPOSSIBLE`, and `traffic_authority: false`. This is a capacity +check, not a tag reservation or proof that a tagged probe is safe. An image-only +plan can remain valid when a tag is impossible; a tagged caller must refuse that +lifecycle. Existing paper and image-only workflows acquire no traffic authority. + +The HK notification canary vendors the QPK operator contract recorded in +`scripts/vendor/cloud-run-tags-source.json`. Its allocation journal is private, +uses the complete configured and observed tag inventory (including zero-percent +entries), and binds the exact target, candidate and operation. A foreign or +unknown occupied tag causes refusal; no alternate tag is searched after a +collision. Short tags use the shared deterministic hash policy, including the +one-character boundary. The canary always refuses nonpositive headroom and never +selects the helper's optional micro-traffic lifecycle. + +Before its first cloud write the canary creates a separate mode-0600 operation +claim next to its original private backup, persists its tag and candidate, and +re-reads both the claim's content and file identity before every cloud mutation, +including restoration. Exclusive creation prevents concurrent or restarted runs +from replaying an uncertain operation, even when backup reuse was requested. +Claims and reservations are intentionally retained after success or failure; +this helper never deletes them or automatically resumes an ambiguous mutation. +Existing configuration, etag, readiness, paused-job and terminal-request gates +remain required. Restore only the owned tag and original scheduler URI after the +existing terminal proof, then verify the complete original distribution and job +configuration. Lost ownership or unverified restore requires operator +reconciliation under separately approved scope. These checks do not make local +ownership and cloud mutations atomic. + ## Plain-DCA Phase 1 consumer binding `application.plain_dca_paper_adapter.consume_plain_dca_preview` consumes the UES diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index f3c977cc..6dfd0ad1 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -3,6 +3,7 @@ import argparse import json import os +import re import sys from collections.abc import Mapping, Sequence from pathlib import Path @@ -46,6 +47,9 @@ def _should_add_local_src(candidate: Path) -> bool: if candidate_str not in sys.path: sys.path.insert(0, candidate_str) +from scripts.vendor.cloud_run_tags import CONTRACT_VERSION, headroom # noqa: E402 + + from strategy_registry import ( # noqa: E402 LONGBRIDGE_PLATFORM, describe_platform_runtime_requirements, @@ -351,16 +355,22 @@ def _build_target_plan( _target_field(target, defaults, "cloud_run_service"), ) runtime_target = _resolve_runtime_target(target, defaults, env, per_service_mode) - if not service_name: - service_name = str(runtime_target.get("service_name") or "").strip() + if service_name is None: + service_name = runtime_target.get("service_name") if not service_name: raise ValueError("Each Cloud Run sync target requires service/service_name") - runtime_target_service = str(runtime_target.get("service_name") or "").strip() - if runtime_target_service and runtime_target_service != service_name: - raise ValueError( - f"Target {service_name} runtime_target.service_name={runtime_target_service!r} does not match" - ) + # Validate original values; never shorten, trim or otherwise rename a target. + if not isinstance(service_name, str) or not re.fullmatch(r"[a-z](?:[a-z0-9-]{0,47}[a-z0-9])?", service_name): + raise ValueError("Cloud Run exact service name is invalid") + runtime_target_service = runtime_target.get("service_name") + if runtime_target_service is not None and ( + not isinstance(runtime_target_service, str) + or not re.fullmatch(r"[a-z](?:[a-z0-9-]{0,47}[a-z0-9])?", runtime_target_service) + ): + raise ValueError("Cloud Run exact runtime service name is invalid") + if runtime_target_service is not None and runtime_target_service != service_name: + raise ValueError("Cloud Run exact service identity does not match") runtime_target["service_name"] = service_name raw_profile = str(runtime_target.get("strategy_profile") or "").strip() @@ -494,6 +504,12 @@ def _build_target_plan( return { "service_name": service_name, + "tag_preflight": { + "contract_version": CONTRACT_VERSION, + "headroom": headroom(service_name), + "tag_status": "AVAILABLE" if headroom(service_name) > 0 else "TAG_IMPOSSIBLE", + "traffic_authority": False, + }, "strategy_profile": canonical_profile, "env": env_values, "scheduler": _build_scheduler_plan( diff --git a/scripts/run_hk_notification_canary.py b/scripts/run_hk_notification_canary.py index 2a8be2e1..0a0653f2 100644 --- a/scripts/run_hk_notification_canary.py +++ b/scripts/run_hk_notification_canary.py @@ -26,6 +26,7 @@ from zoneinfo import ZoneInfo from scripts import record_daily_account_snapshot as snapshot +from scripts.vendor import cloud_run_tags as tags from scripts.verify_deployed_runtime_target_admission import ( APPROVED_HK_PROBE_DIAGNOSTICS_CANDIDATE, _validate_image_only_source, @@ -38,7 +39,7 @@ BASE_APPLICATION_SHA = "922f338fea7c46391b50bb8316ac88154596916f" PROJECT_ID = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") SERVICE_NAME = re.compile(r"^[a-z][a-z0-9-]{0,61}[a-z0-9]$") -TAG_NAME = re.compile(r"^[a-z][a-z0-9-]{0,48}[a-z0-9]$") +TAG_NAME = re.compile(r"^[a-z](?:[a-z0-9-]*[a-z0-9])?$") GIT_SHA = re.compile(r"^[0-9a-f]{40}$") MAX_CLOUD_BODY_BYTES = 1_000_000 MAX_LOG_BODY_BYTES = 256_000 @@ -313,9 +314,10 @@ def _open_directory_nofollow(path: Path) -> int: raise -def _create_backup_file(path: Path, body: bytes, *, root: Path) -> None: +def _create_backup_file(path: Path, body: bytes, *, root: Path) -> tuple[int, int]: + relative_path = path.relative_to(root) if path.is_relative_to(root) else None if ( - path != root / BACKUP_RELATIVE_PATH + relative_path not in {BACKUP_RELATIVE_PATH, BACKUP_RELATIVE_PATH.with_suffix(".operation.json")} or len(body) > MAX_CLOUD_BODY_BYTES or not body ): @@ -331,7 +333,7 @@ def _create_backup_file(path: Path, body: bytes, *, root: Path) -> None: directory_flags = ( os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) ) - for component in BACKUP_RELATIVE_PATH.parts[:-1]: + for component in relative_path.parts[:-1]: created = False try: os.mkdir(component, mode=0o700, dir_fd=directory_descriptor) @@ -351,7 +353,7 @@ def _create_backup_file(path: Path, body: bytes, *, root: Path) -> None: file_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) file_descriptor = os.open( - BACKUP_RELATIVE_PATH.name, + relative_path.name, file_flags, 0o600, dir_fd=directory_descriptor, @@ -370,7 +372,7 @@ def _create_backup_file(path: Path, body: bytes, *, root: Path) -> None: file_descriptor = -1 read_descriptor = os.open( - BACKUP_RELATIVE_PATH.name, + relative_path.name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=directory_descriptor, ) @@ -384,6 +386,7 @@ def _create_backup_file(path: Path, body: bytes, *, root: Path) -> None: if backup_file.read(MAX_CLOUD_BODY_BYTES + 1) != body: raise _stop("backup_readback_mismatch") os.fsync(directory_descriptor) + return read_stat.st_dev, read_stat.st_ino except CanaryError: raise except Exception: @@ -1000,12 +1003,61 @@ def _create_backup( _create_backup_file(BACKUP_PATH, _canonical(payload), root=BACKUP_ROOT) -def _canary_tag(service: str) -> str: - # Cloud Run limits the combined service and traffic-tag length to 46. - suffix_length = min(12, 46 - len(service) - len("hk-")) - if suffix_length < 6: - raise _stop("service_tag_name_budget_invalid") - return "hk-" + uuid.uuid4().hex[:suffix_length] +def _create_operation_claim(body: bytes) -> tuple[int, int]: + """Create once before writes; never clear after an uncertain result or restart.""" + return _create_backup_file(BACKUP_PATH.with_suffix(".operation.json"), body, root=BACKUP_ROOT) + + +def _require_operation_claim(body: bytes, identity: tuple[int, int]) -> None: + """Re-read private ownership before each write, including cleanup.""" + parent = descriptor = -1 + try: + parent = _open_directory_nofollow(BACKUP_PATH.parent) + descriptor = os.open( + BACKUP_PATH.with_suffix(".operation.json").name, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent, + ) + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600 + or info.st_nlink != 1 or (info.st_dev, info.st_ino) != identity + ): + raise ValueError("invalid claim") + with os.fdopen(os.dup(descriptor), "rb") as source: + if source.read(MAX_CLOUD_BODY_BYTES + 1) != body: + raise ValueError("changed claim") + except Exception: + raise _stop("operation_ownership_lost") from None + finally: + for fd in (descriptor, parent): + if fd >= 0: + os.close(fd) + + + +def _canary_tag(service: str, *, operation_id: str, existing_tags: Mapping[str, str | None], + project: str, region: str, candidate_id: str, state_path: Path) -> str: + result = tags.allocate_tag( + service, "hk-canary", existing_tags=existing_tags, operation_id=operation_id, + project=project, region=region, candidate_id=candidate_id, state_path=state_path, + ) + if isinstance(result, tags.Refusal): + raise _stop("tag_allocation_refused") + return result.tag + + +def _tag_inventory(service: Mapping[str, Any]) -> dict[str, str | None]: + inventory: dict[str, str | None] = {} + # Include configured AND observed entries, including zero-percent tags. + for row in _configured_traffic(service) + _traffic_rows(service): + tag = row.get("tag") + if not tag: + continue + revision = row["revision"] + if tag in inventory and inventory[tag] != revision: + raise _stop("tag_inventory_conflict") + inventory[tag] = revision + return inventory def _job_run_status(job: Mapping[str, Any], *, started_at: datetime) -> int | None: @@ -1369,6 +1421,10 @@ def run_hk_notification_canary( ) or not GIT_SHA.fullmatch(BASE_APPLICATION_SHA): raise _stop("source_pin_invalid") + if tags.headroom(service_name) <= 0: + # This canary never opts into a micro-split, even if a flag is present. + raise _stop("tag_allocation_refused") + operation_id = uuid.uuid4().hex session = session_factory() service_resource = _service_resource(project, region, service_name) probe_resource = "" @@ -1377,7 +1433,7 @@ def run_hk_notification_canary( candidate_revision: Mapping[str, Any] | None = None selected_jobs: dict[str, Mapping[str, Any]] = {} traffic_before: list[dict[str, Any]] = [] - tag = _canary_tag(service_name) + claim_body = b"" probe_resumed = False pause_attempted = False try: @@ -1437,8 +1493,6 @@ def run_hk_notification_canary( env={"WORKFLOW_TARGET": "HK"}, project=project, ) - if not TAG_NAME.fullmatch(tag): - raise _stop("tag_invalid") # Detect state drift since the preflight read before starting writes. service_recheck = _service_get(session, service_resource) @@ -1488,12 +1542,23 @@ def run_hk_notification_canary( ) candidate_revision_name = _revision_name(candidate_revision) + tag = _canary_tag( + service_name, operation_id=operation_id, existing_tags=_tag_inventory(service_recheck), + project=project, region=region, candidate_id=candidate_revision_name, + state_path=BACKUP_PATH.with_suffix(".tags.json"), + ) + claim_body = _canonical({ + "contract_version": tags.CONTRACT_VERSION, "operation_id": operation_id, + "target": service_resource, "candidate": candidate_revision_name, "tag": tag, + }) + claim_identity = _create_operation_claim(claim_body) traffic_with_tag = _tagged_traffic( traffic_before, revision=candidate_revision_name, tag=tag ) etag = str(original_service.get("etag") or "") if not etag: raise _stop("service_etag_missing") + _require_operation_claim(claim_body, claim_identity) try: _service_patch_traffic( session, service_resource, etag=etag, traffic=traffic_with_tag @@ -1521,6 +1586,7 @@ def run_hk_notification_canary( original_job, resource=probe_resource, base_uri=contract["base_uri"] ) temporary_probe_uri = f"{tagged_url}/probe" + _require_operation_claim(claim_body, claim_identity) try: _scheduler_patch_uri(session, probe_resource, temporary_probe_uri) except Exception: @@ -1538,6 +1604,7 @@ def pause_once() -> None: if pause_attempted: return pause_attempted = True + _require_operation_claim(claim_body, claim_identity) try: snapshot._scheduler_action(session, job_config, "pause") except Exception: @@ -1556,6 +1623,7 @@ def pause_once() -> None: raise _stop("probe_job_temporary_readback_failed") original_last_attempt = original_job.get("lastAttemptTime") + _require_operation_claim(claim_body, claim_identity) try: snapshot._scheduler_action( session, @@ -1594,6 +1662,7 @@ def pause_once() -> None: ): pause_once() raise _stop("scheduler_natural_window") + _require_operation_claim(claim_body, claim_identity) try: snapshot._scheduler_run(session, job_config) except Exception: @@ -1633,6 +1702,7 @@ def pause_once() -> None: del request_status # Restore only our modified URI, then remove only our unique tag. + _require_operation_claim(claim_body, claim_identity) try: _scheduler_patch_uri(session, probe_resource, base_probe_uri) except Exception: @@ -1657,6 +1727,7 @@ def pause_once() -> None: final_traffic = _remove_owned_tag( current_service, revision=candidate_revision_name, tag=tag ) + _require_operation_claim(claim_body, claim_identity) try: _service_patch_traffic( session, diff --git a/scripts/vendor/__init__.py b/scripts/vendor/__init__.py new file mode 100644 index 00000000..ddcf215d --- /dev/null +++ b/scripts/vendor/__init__.py @@ -0,0 +1 @@ +"""Pinned operator contracts; see cloud-run-tags-source.json.""" diff --git a/scripts/vendor/cloud-run-tags-source.json b/scripts/vendor/cloud-run-tags-source.json new file mode 100644 index 00000000..32529e26 --- /dev/null +++ b/scripts/vendor/cloud-run-tags-source.json @@ -0,0 +1,8 @@ +{ + "contract_version": "cloud-run-tags/v1", + "source_repository": "QuantStrategyLab/QuantPlatformKit", + "source_path": "src/quant_platform_kit/operators/cloud_run_tags.py", + "source_job": "qpk-cr-tag-helper-r1", + "source_status": "unmerged implementation snapshot", + "sha256": "ecb8762ddd8e76c1fb6bf4de2981347951ba9640442632b1e3aaa459d5fc8c05" +} diff --git a/scripts/vendor/cloud_run_tags.py b/scripts/vendor/cloud_run_tags.py new file mode 100644 index 00000000..7001cf51 --- /dev/null +++ b/scripts/vendor/cloud_run_tags.py @@ -0,0 +1,228 @@ +"""Versioned offline Cloud Run tag policy. This module never calls a cloud API. + +Allocation is an exclusive local reservation, NOT permission to mutate a service. +See docs/cloud-run-tags.md for inventory, ownership and restore requirements. +""" +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import re +import stat +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path + +CONTRACT_VERSION = 'cloud-run-tags/v1' +TAGGED_URL_BUDGET = 46 +_NAME = re.compile(r'[a-z](?:[a-z0-9-]*[a-z0-9])?\Z') +_DIGEST = re.compile(r'[0-9a-f]{64}\Z') + + +@dataclass(frozen=True) +class Refusal: + """Explicit failure with a static, identifier-free reason code.""" + reason: str + + +@dataclass(frozen=True) +class AllocatedTag: + """Operational value: do not log the tag or serialize it into public output.""" + tag: str = field(repr=False) + + +@dataclass(frozen=True) +class Target: + project: str = field(repr=False) + region: str = field(repr=False) + service_name: str = field(repr=False) + + +@dataclass(frozen=True) +class WarmupApproval: + """Caller-verified owner decision; this value cannot authenticate an approval.""" + target: Target = field(repr=False) + action: str = field(repr=False) + operation_id: str = field(repr=False) + warmup_traffic_approved: bool = field(repr=False) + + +@dataclass(frozen=True) +class Lifecycle: + kind: str + reason: str + previous_percent: int = 0 + candidate_percent: int = 0 + + +def headroom(service_name: str) -> int: + """Measure the exact original name, before any display masking.""" + return TAGGED_URL_BUDGET - len(service_name) + + +def _nonempty(value: object) -> bool: + return isinstance(value, str) and bool(value.strip()) and value == value.strip() + + +def _name(value: object) -> bool: + return isinstance(value, str) and bool(_NAME.fullmatch(value)) + + +def _target_valid(target: object) -> bool: + return (isinstance(target, Target) and _nonempty(target.project) + and _nonempty(target.region) and _name(target.service_name)) + + +def lifecycle_for_impossible_tag( + target: Target, action: str, operation_id: str, *, + approval: WarmupApproval | None = None, no_traffic: bool = True, + image_only: bool = False, +) -> Lifecycle: + """Default STOP; permit only an exact, explicitly approved warmup 99/1 plan. + + A returned plan still requires the caller's exclusive, single-use execution + claim and fresh cloud readback. It never grants deployment/traffic authority. + """ + if not _target_valid(target) or not _nonempty(operation_id): + return Lifecycle('STOP', 'INVALID_INPUT') + if no_traffic is not False or image_only is not False: + return Lifecycle('STOP', 'NO_TRAFFIC_REQUIRED') + if headroom(target.service_name) > 0: + return Lifecycle('STOP', 'TAG_NOT_IMPOSSIBLE') + if (action != 'warmup' or not isinstance(approval, WarmupApproval) + or approval.warmup_traffic_approved is not True + or approval.target != target or approval.action != action + or approval.operation_id != operation_id): + return Lifecycle('STOP', 'EXACT_WARMUP_APPROVAL_REQUIRED') + return Lifecycle('MICRO_SPLIT', 'EXACT_WARMUP_APPROVAL', 99, 1) + + +def _digest(*parts: str) -> str: + return hashlib.sha256(json.dumps(parts, separators=(',', ':')).encode()).hexdigest() + + +def _short_tag(scope: str, operation: str, candidate: str, size: int) -> str: + digest = _digest(CONTRACT_VERSION, scope, operation, candidate) + # Always begin with a letter, including the one-character boundary. + return (chr(ord('a') + int(digest[:2], 16) % 26) + digest[2:])[:size] + + +def _journal_valid(data: object) -> bool: + if (not isinstance(data, dict) or set(data) != {'contract_version', 'scopes'} + or data['contract_version'] != CONTRACT_VERSION + or not isinstance(data['scopes'], dict)): + return False + for scope, records in data['scopes'].items(): + if (not isinstance(scope, str) or not _DIGEST.fullmatch(scope) + or not isinstance(records, dict)): + return False + tags = set() + for operation, record in records.items(): + if (not isinstance(operation, str) or not _DIGEST.fullmatch(operation) + or not isinstance(record, dict) + or set(record) != {'candidate', 'preferred', 'tag'} + or any(not isinstance(record[k], str) or not _DIGEST.fullmatch(record[k]) + for k in ('candidate', 'preferred')) + or not _name(record['tag']) or record['tag'] in tags): + return False + tags.add(record['tag']) + return True + + +def _same_file(path: Path, fd: int) -> bool: + current = os.stat(path, follow_symlinks=False) + held = os.fstat(fd) + return (stat.S_ISREG(current.st_mode) and current.st_nlink == 1 + and (current.st_dev, current.st_ino) == (held.st_dev, held.st_ino) + and stat.S_IMODE(current.st_mode) == 0o600) + + +def _reserve(data: dict, scope: str, operation: str, candidate: str, + preferred: str, tag: str, existing: Mapping) -> AllocatedTag | Refusal: + records = data['scopes'].setdefault(scope, {}) + record = {'candidate': candidate, 'preferred': preferred, 'tag': tag} + previous = records.get(operation) + if previous is not None and previous != record: + return Refusal('OPERATION_REBOUND') + if any(key != operation and value['tag'] == tag for key, value in records.items()): + return Refusal('TAG_RESERVED') + # A cloud revision match alone does not prove operation ownership. + if tag in existing and (previous != record or _digest(existing[tag]) != candidate): + return Refusal('TAG_OCCUPIED') + records[operation] = record + return AllocatedTag(tag) + + +def allocate_tag( + service_name: str, preferred_tag: str, *, existing_tags: Mapping[str, str | None], + operation_id: str, candidate_id: str, state_path: str | Path, + project: str, region: str, +) -> AllocatedTag | Refusal: + """Reserve preferred or one deterministic shortened tag; never overwrite. + + existing_tags must come from a fresh, complete inventory (including zero- + percent entries); None as a value is occupied with unknown ownership. Use + one private, shared job-local state_path for all processes/restarts. Only + the allocator creates this file. Missing evidence or state I/O fails closed. + """ + if (not _target_valid(Target(project, region, service_name)) + or not _name(preferred_tag) or not _nonempty(operation_id) + or not _nonempty(candidate_id) or not isinstance(existing_tags, Mapping) + or any(not _name(k) or (v is not None and not _nonempty(v)) + for k, v in existing_tags.items())): + return Refusal('INVALID_INPUT') + room = headroom(service_name) + if room <= 0: + return Refusal('TAG_IMPOSSIBLE') + scope = _digest(project, region, service_name) + operation, candidate, preferred = map(_digest, (operation_id, candidate_id, preferred_tag)) + tag = preferred_tag if len(preferred_tag) <= room else _short_tag(scope, operation, candidate, room) + # Snapshot caller input without ever mutating it. Cloud freshness remains a + # caller obligation; the reservation ledger is read only AFTER flock. + existing = dict(existing_tags) + fd = None + try: + path = Path(state_path) + try: + fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + created = True + except FileExistsError: + fd = os.open(path, os.O_RDWR | os.O_NOFOLLOW) + created = False + with os.fdopen(fd, 'r+', encoding='utf-8') as stream: + fd = None + fcntl.flock(stream.fileno(), fcntl.LOCK_EX) + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + text = stream.read() + # Another process may have initialized a just-created file first. + data = json.loads(text) if text else ({'contract_version': CONTRACT_VERSION, 'scopes': {}} if created else None) + if not _journal_valid(data): + return Refusal('STATE_UNAVAILABLE') + result = _reserve(data, scope, operation, candidate, preferred, tag, existing) + if isinstance(result, Refusal): + return result + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + stream.seek(0) + json.dump(data, stream, sort_keys=True, separators=(',', ':')) + stream.truncate() + stream.flush() + os.fsync(stream.fileno()) + # Persist the directory entry before returning a usable allocation. + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + return result + except (OSError, ValueError, TypeError, RecursionError): + # Do not include exception text: paths/resources may identify accounts. + return Refusal('STATE_UNAVAILABLE') + finally: + if fd is not None: + os.close(fd) diff --git a/tests/test_cr_tag_adoption.py b/tests/test_cr_tag_adoption.py new file mode 100644 index 00000000..a2356b69 --- /dev/null +++ b/tests/test_cr_tag_adoption.py @@ -0,0 +1,130 @@ +"""Offline caller regressions; all target identities are synthetic.""" +from __future__ import annotations +import json +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path +import pytest +from scripts import build_cloud_run_env_sync_plan as planner +from scripts import run_hk_notification_canary as canary + + +def _plan(service, runtime_service=None): + return planner.build_sync_plan({ + "PLATFORM_CONFIG_JSON": "{}", + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps({ + "defaults": {"GLOBAL_TELEGRAM_CHAT_ID": "fake-chat", "NOTIFY_LANG": "en"}, + "targets": [{"service": service, "account_prefix": "PAPER", "runtime_target": { + "platform_id": "longbridge", "strategy_profile": "tqqq_growth_income", + "deployment_selector": "PAPER", "account_scope": "PAPER", + "account_selector": ["PAPER"], "execution_mode": "paper", "dry_run_only": True, + "service_name": service if runtime_service is None else runtime_service, + }}], + }), + })["targets"][0] + + +@pytest.mark.parametrize("service", ["invalid/name", " x", "UPPER", "x" * 50, 0, False]) +def test_plan_rejects_exact_invalid_service_without_echo(service): + with pytest.raises(ValueError) as error: + _plan(service) + assert str(service) not in str(error.value) + + +@pytest.mark.parametrize("runtime_service", [" fake-service ", 0, False]) +def test_plan_rejects_runtime_service_whitespace_instead_of_renaming(runtime_service): + with pytest.raises(ValueError): + _plan("fake-service", runtime_service) + + +@pytest.mark.parametrize("length", [20, 34, 45, 46, 48]) +def test_plan_reports_budget_without_granting_tag_or_traffic_authority(length): + target = _plan("s" * length) + preflight = target["tag_preflight"] + assert preflight["headroom"] == 46 - length + assert preflight["tag_status"] == ("AVAILABLE" if length < 46 else "TAG_IMPOSSIBLE") + assert preflight["traffic_authority"] is False + assert target["service_name"] == "s" * length + + +@pytest.mark.parametrize("room", range(1, 13)) +def test_canary_allocates_short_tag_from_exact_candidate_and_operation(room, tmp_path): + service = "s" * (46 - room) + options = dict(operation_id="fake-run", project="fake-project", region="fake-region", candidate_id="fake-candidate", state_path=tmp_path / "tags.json") + tag = canary._canary_tag(service, existing_tags={}, **options) + assert 1 <= len(tag) <= room + assert tag[0].islower() + assert tag == canary._canary_tag(service, existing_tags={}, **options) + with pytest.raises(canary.CanaryError, match="tag_allocation_refused"): + canary._canary_tag(service, existing_tags={tag: None}, **options) + + +def test_canary_impossible_budget_refuses_before_opening_cloud_session(): + opened = [] + with pytest.raises(canary.CanaryError, match="tag_allocation_refused"): + canary.run_hk_notification_canary(env={ + "WORKFLOW_TARGET": "HK", "GITHUB_REF": "refs/heads/main", + "GITHUB_REPOSITORY": "QuantStrategyLab/LongBridgePlatform", + "GOOGLE_CLOUD_PROJECT": "fake-project", "CLOUD_RUN_REGION": "asia-east2", + "CLOUD_RUN_SERVICE": "s" * 48, "HK_CANDIDATE_REVISION": "fake-candidate", + "HK_NOTIFICATION_CANARY_BACKUP_PATH": str(canary.BACKUP_PATH), + "WARMUP_TRAFFIC_APPROVED": "true", + }, session_factory=lambda: opened.append(True)) + assert opened == [] + + +def _claim_worker(root): + canary.BACKUP_ROOT = Path(root) + canary.BACKUP_PATH = Path(root) / canary.BACKUP_RELATIVE_PATH + try: + canary._create_operation_claim(b'{"operation_id":"fake-run"}') + except canary.CanaryError: + return False + return True + + +def test_operation_claim_is_exclusive_across_processes_and_restart(tmp_path): + with ProcessPoolExecutor(max_workers=2) as pool: + assert sorted(pool.map(_claim_worker, [str(tmp_path)] * 2)) == [False, True] + assert _claim_worker(str(tmp_path)) is False + + +def test_operation_claim_rejects_changed_or_missing_owner(tmp_path, monkeypatch): + monkeypatch.setattr(canary, "BACKUP_ROOT", tmp_path) + monkeypatch.setattr(canary, "BACKUP_PATH", tmp_path / canary.BACKUP_RELATIVE_PATH) + body = b'{"operation_id":"fake-run"}' + identity = canary._create_operation_claim(body) + canary._require_operation_claim(body, identity) + claim = canary.BACKUP_PATH.with_suffix(".operation.json") + claim.write_bytes(b'{"operation_id":"fake-foreign-run"}') + with pytest.raises(canary.CanaryError, match="operation_ownership_lost"): + canary._require_operation_claim(body, identity) + claim.unlink() + with pytest.raises(canary.CanaryError, match="operation_ownership_lost"): + canary._require_operation_claim(body, identity) + + +def test_operation_claim_replacement_with_identical_bytes_loses_ownership(tmp_path, monkeypatch): + monkeypatch.setattr(canary, "BACKUP_ROOT", tmp_path) + monkeypatch.setattr(canary, "BACKUP_PATH", tmp_path / canary.BACKUP_RELATIVE_PATH) + body = b'{"operation_id":"fake-run"}' + identity = canary._create_operation_claim(body) + path = canary.BACKUP_PATH.with_suffix(".operation.json") + replacement = path.with_suffix(".replacement") + replacement.write_bytes(body) + replacement.chmod(0o600) + replacement.replace(path) + with pytest.raises(canary.CanaryError, match="operation_ownership_lost"): + canary._require_operation_claim(body, identity) + + +def test_plan_service_identity_mismatch_does_not_echo_names(): + with pytest.raises(ValueError) as error: + _plan("fake-sensitive-service", "fake-other-service") + assert "fake-sensitive-service" not in str(error.value) + assert "fake-other-service" not in str(error.value) + + +@pytest.mark.parametrize("service", [0, False]) +def test_plan_invalid_configured_service_cannot_fall_back_to_runtime(service): + with pytest.raises(ValueError): + _plan(service, "fake-service") diff --git a/tests/test_hk_notification_canary.py b/tests/test_hk_notification_canary.py index 95636a8f..bbb3d4a6 100644 --- a/tests/test_hk_notification_canary.py +++ b/tests/test_hk_notification_canary.py @@ -863,14 +863,20 @@ def test_service_config_ignores_only_output_modifier_and_keeps_config_changes(): assert canary._service_config_view(current) != canary._service_config_view(original) -def test_canary_tag_respects_provider_combined_name_limit(): - for length in (20, 31, 36, 37): +def test_canary_tag_respects_provider_combined_name_limit(tmp_path): + for length in (20, 31, 36, 37, 38, 45): name = "s" * length - tag = canary._canary_tag(name) + tag = canary._canary_tag( + name, operation_id="fake-run", existing_tags={}, project="fake-project", + region="fake-region", candidate_id="fake-candidate", state_path=tmp_path / "tags.json", + ) assert len(name) + len(tag) <= 46 - assert tag.startswith("hk-") and len(tag) >= 9 - with pytest.raises(canary.CanaryError, match="service_tag_name_budget_invalid"): - canary._canary_tag("s" * 38) + assert canary.TAG_NAME.fullmatch(tag) + with pytest.raises(canary.CanaryError, match="tag_allocation_refused"): + canary._canary_tag( + "s" * 46, operation_id="fake-run", existing_tags={}, project="fake-project", + region="fake-region", candidate_id="fake-candidate", state_path=tmp_path / "tags.json", + ) def test_backup_reuse_is_read_only_and_requires_exact_original_snapshot( @@ -1140,3 +1146,40 @@ def test_retired_target_inspection_preserves_disabled_paper_guards(profile, unsa env["RUNTIME_TARGET_JSON"] = json.dumps(target) with pytest.raises(canary.CanaryError, match="runtime_target_mismatch"): canary._runtime_target(env, service=SERVICE) + + +def test_canary_lost_claim_blocks_first_cloud_write(monkeypatch, tmp_path): + run, state = _run_synthetic_canary_with_delayed_dispatch(monkeypatch, tmp_path) + create = canary._create_operation_claim + + def steal(body): + identity = create(body) + canary.BACKUP_PATH.with_suffix(".operation.json").write_bytes(b"{}") + return identity + + monkeypatch.setattr(canary, "_create_operation_claim", steal) + with pytest.raises(canary.CanaryError, match="operation_ownership_lost"): + run() + assert state["events"] == [] + + +def test_canary_claim_survives_success_and_blocks_restart(monkeypatch, tmp_path): + run, state = _run_synthetic_canary_with_delayed_dispatch(monkeypatch, tmp_path) + assert run() == "completed" + events_before = list(state["events"]) + claim = json.loads(canary.BACKUP_PATH.with_suffix(".operation.json").read_bytes()) + assert claim["contract_version"] == "cloud-run-tags/v1" + assert claim["candidate"] == CANDIDATE + with pytest.raises(canary.CanaryError): + run() + assert state["events"] == events_before + + +def test_tag_inventory_preserves_zero_percent_and_rejects_new_conflict(): + service = _service() + service["traffic"].append({"revision": "fake-old", "tag": "fake-zero", "percent": 0}) + service["trafficStatuses"].append({"revision": "fake-old", "tag": "fake-zero", "percent": 0}) + assert canary._tag_inventory(service) == {"fake-zero": "fake-old"} + service["trafficStatuses"][-1]["revision"] = "fake-foreign" + with pytest.raises(canary.CanaryError, match="tag_inventory_conflict"): + canary._tag_inventory(service) diff --git a/tests/test_vendor_cr_tag_contract.py b/tests/test_vendor_cr_tag_contract.py new file mode 100644 index 00000000..32273d5a --- /dev/null +++ b/tests/test_vendor_cr_tag_contract.py @@ -0,0 +1,194 @@ +"""Offline contract tests; all resource and operation identities are synthetic.""" +from __future__ import annotations + +import importlib +import json +import multiprocessing +from pathlib import Path + +import pytest + + +def api(): + return importlib.import_module('scripts.vendor.cloud_run_tags') + + +def allocate(tmp_path, **overrides): + options = dict( + service_name='example-service', preferred_tag='cand-abcdef01', + existing_tags={}, operation_id='fake-run-one', candidate_id='fake-revision-one', + project='', region='example-region', state_path=tmp_path / 'tags.json', + ) + options.update(overrides) + return api().allocate_tag(**options) + + +def test_preferred_tag_and_exact_headroom(tmp_path): + result = allocate(tmp_path) + assert isinstance(result, api().AllocatedTag) + assert result.tag == 'cand-abcdef01' + assert api().headroom('example-service') == 31 + assert api().headroom('x' * 33) == 13 + + +@pytest.mark.parametrize('room', range(1, 13)) +def test_shorten_every_tight_budget_and_recover_after_restart(tmp_path, room): + result = allocate(tmp_path, service_name='x' * (46 - room)) + assert isinstance(result, api().AllocatedTag) + assert 1 <= len(result.tag) <= room + assert result.tag[0].isalpha() and result.tag.isalnum() + assert allocate(tmp_path, service_name='x' * (46 - room)) == result + assert json.loads((tmp_path / 'tags.json').read_text())['contract_version'] == api().CONTRACT_VERSION + + +@pytest.mark.parametrize('length', [46, 47, 48, 49, 63]) +def test_impossible_never_returns_silent_success(tmp_path, length): + result = allocate(tmp_path, service_name='x' * length) + assert result == api().Refusal('TAG_IMPOSSIBLE') + assert not (tmp_path / 'tags.json').exists() + + +@pytest.mark.parametrize('occupant', ['foreign-revision', None]) +def test_foreign_or_unknown_preferred_tag_refused(tmp_path, occupant): + assert allocate(tmp_path, existing_tags={'cand-abcdef01': occupant}) == api().Refusal('TAG_OCCUPIED') + + +def test_cloud_revision_match_without_local_ownership_is_not_owned(tmp_path): + assert allocate(tmp_path, existing_tags={'cand-abcdef01': 'fake-revision-one'}) == api().Refusal('TAG_OCCUPIED') + + +def test_owned_tag_and_zero_percent_tags_are_preserved(tmp_path): + allocated = allocate(tmp_path) + # Tag inventories include ALL entries, including zero-percent tags. + tags = {allocated.tag: 'fake-revision-one', 'zero-percent': 'fake-previous'} + before = dict(tags) + assert allocate(tmp_path, existing_tags=tags) == allocated + assert tags == before + assert allocate(tmp_path, existing_tags={allocated.tag: 'foreign-revision'}) == api().Refusal('TAG_OCCUPIED') + + +def test_hash_collision_refused_without_searching_another_tag(tmp_path): + first = allocate(tmp_path, service_name='x' * 45) + assert allocate(tmp_path, service_name='x' * 45, existing_tags={first.tag: None}) == api().Refusal('TAG_OCCUPIED') + + +def test_operation_identity_cannot_be_rebound_after_restart(tmp_path): + allocate(tmp_path) + assert allocate(tmp_path, candidate_id='fake-revision-two') == api().Refusal('OPERATION_REBOUND') + assert allocate(tmp_path, preferred_tag='other') == api().Refusal('OPERATION_REBOUND') + + +def test_latest_disk_reservation_wins_over_free_cloud_snapshot(tmp_path): + allocate(tmp_path) + assert allocate(tmp_path, operation_id='fake-run-two') == api().Refusal('TAG_RESERVED') + + +def test_missing_or_corrupt_inventory_and_journal_fail_closed(tmp_path): + assert allocate(tmp_path, existing_tags=None) == api().Refusal('INVALID_INPUT') + path = tmp_path / 'tags.json' + path.write_text('{incomplete') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + path.write_text('{}') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + + +def test_truncated_existing_journal_is_not_reinitialized(tmp_path): + allocate(tmp_path) + (tmp_path / 'tags.json').write_text('') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + + +def test_state_does_not_disclose_resource_or_candidate_identity(tmp_path): + result = allocate(tmp_path, service_name='fake-sensitive-service', candidate_id='fake-sensitive-candidate', operation_id='fake-sensitive-run') + data = (tmp_path / 'tags.json').read_text() + for value in ['fake-sensitive-service', 'fake-sensitive-candidate', 'fake-sensitive-run', '', 'example-region']: + assert value not in data + assert value not in repr(result) + target = api().Target('', 'example-region', 'fake-sensitive-service') + assert 'fake-sensitive-service' not in repr(target) + assert 'fake-sensitive-run' not in repr(api().WarmupApproval(target, 'warmup', 'fake-sensitive-run', True)) + + +def _allocate_process(path, run, queue): + result = allocate(Path(path), operation_id=run) + queue.put(type(result).__name__) + + +def test_two_processes_cannot_reserve_same_free_tag(tmp_path): + ctx = multiprocessing.get_context('spawn') + queue = ctx.Queue() + workers = [ctx.Process(target=_allocate_process, args=(str(tmp_path), f'fake-run-{n}', queue)) for n in range(2)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(15) + assert worker.exitcode == 0 + assert sorted(queue.get(timeout=2) for _ in workers) == ['AllocatedTag', 'Refusal'] + + +def test_symlink_state_is_refused(tmp_path): + other = tmp_path / 'other.json' + other.write_text('') + (tmp_path / 'tags.json').symlink_to(other) + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + assert other.read_text() == '' + + +@pytest.mark.parametrize('changes', [ + {'service_name': ' uppercase'}, {'preferred_tag': ''}, {'preferred_tag': '9bad'}, + {'operation_id': ''}, {'candidate_id': ''}, {'region': ''}, {'project': ''}, + {'existing_tags': {'bad tag': None}}, {'existing_tags': {'valid': 7}}, +]) +def test_invalid_inputs_refused(tmp_path, changes): + assert allocate(tmp_path, **changes) == api().Refusal('INVALID_INPUT') + + +def lifecycle(**overrides): + module = api() + target = module.Target('', 'example-region', 'x' * 48) + options = dict(target=target, action='warmup', operation_id='fake-run', no_traffic=False) + options.update(overrides) + return module.lifecycle_for_impossible_tag(**options) + + +def approval(**overrides): + module = api() + options = dict(target=module.Target('', 'example-region', 'x' * 48), action='warmup', operation_id='fake-run', warmup_traffic_approved=True) + options.update(overrides) + return module.WarmupApproval(**options) + + +def test_impossible_lifecycle_stops_by_default(): + assert lifecycle().kind == 'STOP' + assert lifecycle().candidate_percent == 0 + + +def test_only_exact_explicit_approval_selects_micro_split(): + result = lifecycle(approval=approval()) + assert result.kind == 'MICRO_SPLIT' + assert (result.previous_percent, result.candidate_percent) == (99, 1) + + +@pytest.mark.parametrize('changes', [ + {'target': lambda m: m.Target('', 'example-region', 'different-service')}, + {'target': lambda m: m.Target('', 'different-region', 'x' * 48)}, + {'target': lambda m: m.Target('fake-other-project', 'example-region', 'x' * 48)}, + {'action': lambda m: 'promotion'}, {'operation_id': lambda m: 'another-run'}, + {'warmup_traffic_approved': lambda m: False}, + {'warmup_traffic_approved': lambda m: 'true'}, +]) +def test_approval_does_not_widen(changes): + assert lifecycle(approval=approval(**{key: value(api()) for key, value in changes.items()})).kind == 'STOP' + + +@pytest.mark.parametrize('changes', [{'no_traffic': True}, {'image_only': True}, {'no_traffic': None}, {'image_only': None}]) +def test_no_traffic_and_image_only_never_select_micro_split(changes): + result = lifecycle(approval=approval(), **changes) + assert result.kind == 'STOP' + assert result.candidate_percent == 0 + + +def test_fitting_service_and_other_action_cannot_use_fallback(): + target = api().Target('', 'example-region', 'short-service') + assert lifecycle(target=target, approval=approval(target=target)).kind == 'STOP' + assert lifecycle(action='promotion', approval=approval(action='promotion')).kind == 'STOP'