From f8d83da0605ac22593b19db3a7cd0615116af9a3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:57:16 +0800 Subject: [PATCH 1/5] fix: Add one-shot retained-image warm recovery Co-Authored-By: Codex --- docs/ibkr-runtime-rollout.md | 99 +++++++ scripts/cloud_run_warm_recovery.py | 364 ++++++++++++++++++++++++++ tests/test_cloud_run_warm_recovery.py | 257 ++++++++++++++++++ 3 files changed, 720 insertions(+) create mode 100644 scripts/cloud_run_warm_recovery.py create mode 100644 tests/test_cloud_run_warm_recovery.py diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index 50d785d6..37966b36 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -567,3 +567,102 @@ 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. + +### One-shot warm-baseline recovery entry + +`scripts/cloud_run_warm_recovery.py` is a separate retained-image entry. It does +not change the ordinary protected build workflow or its minimum-zero baseline +rule. Its only baseline exception is the exact owner-card-bound warm revision at +100% traffic and revision minimum one, with service and template minima zero. +It creates **one new replacement** at revision minimum zero from the existing +service template and the approved immutable image. There is no application +checkout, build or image push. A missing digest stops before mutation; a build +requires a separate scope and the ordinary image workflow. + +Run the operator code at the newly approved recovery merge, in the existing +protected execution environment after that environment's approval. Source merge, +`agent-approved`, a card file, and the `RECOVERY_PROTECTED_ENVIRONMENT` variable +are not operational approval. The variable binds the runner's environment name; +it does not implement or bypass the hosting platform's environment gate. No new +workflow dispatch is introduced. A follow-up owner-approved operator job invokes: + +```bash +RECOVERY_PROTECTED_ENVIRONMENT='' \ + uv run --no-sync python -m scripts.cloud_run_warm_recovery \ + --card /protected/recovery-card.json --attempt /protected/unique-attempt.json +``` + +The card is a mode-0600 JSON file supplied through approved protected storage, +never a PR, command argument, log or public artifact. Keep its contents and the +attempt outside Git. The source checkout must have the card's exact full +`operator_commit` and no tracked modifications. Use a fresh request/attempt +identity once; do not rerun or delete an attempt file to resume. The full original +service snapshot is also a precondition: a different remote state cannot resume +an earlier operation. Any later attempt needs a fresh owner card and readback. + +| Card field | Required binding | +| --- | --- | +| `authority`, `protected_environment`, `expires_at` | Explicit owner authority, protected runner environment and timezone-aware expiry, including the permitted minimum-one time/cost window | +| `scope` | Exactly `s3-only`; no other peer or service | +| `project`, `region`, `service` | One full identity from the protected S3 inventory; no wildcard | +| `operator_commit`, `source_commit` | Full operator merge and separate retained application's commit label | +| `image_digest` | Exact approved `sha256:` digest, equal to the warm revision's image digest | +| `request_id`, `replacement` | New lowercase request identity; replacement exactly `-rec-`, at most 63 characters | +| `service_snapshot` | Full fresh Cloud Run v1 service GET, including UID, resourceVersion, generation, metadata, template, spec and status; exact baseline equality is required | +| `warm` | Full fresh v1 warm revision GET, including UID, generation, source/run labels, annotations, full spec and imageDigest | +| `restore_warm_on_failure` | Exactly `true`, explicitly permitting one warm-100 compensation after an activation/promotion failure or unknown outcome, including after forward approval expiry | + +The baseline must use an explicit warm-100 desired traffic row, with no tags or +latest-revision indirection. The existing template must already carry the +approved application source label, a single application container, an untagged +image repository and minimum zero. Configuration and Secret references are +copied without synchronization; runtime target continuity remains checked by +the common verifier. No Secret value, IAM policy, Scheduler, execution switch or +risk setting is changed. + +Execution qualifies the frozen warm twice, five seconds apart. It then creates +the exact replacement without routing traffic to it, shifts warm=99 and +replacement=1, and allows at most 300 seconds for strict startup qualification. +Two fresh healthy observations must be separated by five seconds; Ready, +Active, ContainerReady and ContainerHealthy must each be uniquely True, with +current observed generation and unchanged revision identity/spec/digest. A +Retired or unknown result is never healthy. A final fresh check precedes the +exact replacement=100 update. No application HTTP request is sent, including to +health routes; the entry observes Cloud Run's configured startup-probe evidence. + +All writes use the service's current `metadata.resourceVersion` as an optimistic +concurrency precondition. Only template name/image/minimum and the explicit +traffic rows change; the service UID, desired configuration and metadata remain +bound. Provider-generated operation/URL/creator/status annotations are excluded +from desired metadata comparison. See the +[Cloud Run v1 replaceService contract](https://docs.cloud.google.com/run/docs/reference/rest/v1/namespaces.services/replaceService). +Successful writes allow up to 60 seconds of control-plane convergence, with no +write replay. Health qualification retains its separate 300-second deadline. + +After any activation or promotion failure, including an ambiguous write, the +entry reads the actual state and attempts **one distinct warm=100 compensation**. +The warm's frozen identity/spec/digest must still match the already qualified +baseline. It need not remain Active while temporarily unreferenced. Only the +original warm-100, exact 99/1, or exact replacement-100 distributions are +eligible. Concurrent identity/configuration changes or unknown third-party +traffic stop compensation rather than overwrite another actor. A rejected, +interrupted or ambiguous compensation is never retried. Inspect the private +attempt status; `FAILED_RESTORE_UNCONFIRMED` must not be reported as restored. +A failure before activation never advances traffic or creates another revision. + +Before and after execution, including a readable failure state, independent +[Cloud Run v2 services.get](https://docs.cloud.google.com/run/docs/reference/rest/v2/projects.locations.services/get) +checks service and template minimum zero. The private attempt records these +minima and the warm revision's MinInstancesProvisioned observation. Successful +final serving revision minimum zero is verified separately. An unreferenced +minimum-one revision or zero service minimum does not prove zero billing; +unknown provisioning remains unknown and no billing total is queried. All raw +provider diagnostics and credential tokens remain private and are not printed. + +A later owner deploy card must pin all fields above, the exact reviewed recovery +merge, the retained image/source, fresh complete snapshots, the attempt identity +and expiry. It must explicitly approve one replacement, 99/1, the bounded health +window, replacement-100 and warm-100 compensation, with independent minimum/cost +readbacks. Exclude Peer1/Peer2/Peer4, all other services, Secret/IAM/Scheduler, +environment sync, risk/execution switches, orders, deletion and automatic retry. +Engineering acceptance does not authorize invoking this entry. diff --git a/scripts/cloud_run_warm_recovery.py b/scripts/cloud_run_warm_recovery.py new file mode 100644 index 00000000..76d04e86 --- /dev/null +++ b/scripts/cloud_run_warm_recovery.py @@ -0,0 +1,364 @@ +"""One-shot retained-image recovery; an owner-approved card is required. + +No build, environment sync, health HTTP invocation, or mutation outside one +service. Private cards and attempt files must remain outside the repository. +""" +from __future__ import annotations + +import argparse +import contextlib +import copy +from datetime import datetime, timezone +import io +import json +import os +from pathlib import Path +import re +import signal +import subprocess +import time +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +from scripts.cloud_run_candidate_warmup import MINIMUM, active, revision_metadata +from scripts.runtime_target_continuity import StartupHealthPending, verify_traffic_revision + + +OUTPUT_ANNOTATIONS = {'run.googleapis.com/operation-id', 'run.googleapis.com/urls', + 'run.googleapis.com/ingress-status', 'serving.knative.dev/creator', + 'serving.knative.dev/lastModifier'} + + +def service_meta(value: dict, key: str): + result = value['metadata'].get(key) + if key == 'annotations': + return {name: item for name, item in (result or {}).items() if name not in OUTPUT_ANNOTATIONS} + return result + + +class RecoveryError(RuntimeError): + """A fixed diagnostic with no private provider details.""" + + +def require(test: bool) -> None: + if not test: + raise RecoveryError('recovery binding, continuity or readback mismatch') + + +def fingerprint(value: dict) -> dict: + return {'metadata': revision_metadata(value), 'spec': value['spec'], + 'imageDigest': value.get('status', {}).get('imageDigest')} + + +def healthy(value: dict, commit: str) -> None: + # The common verifier prints provider conditions; keep those private here. + with contextlib.redirect_stderr(io.StringIO()), contextlib.redirect_stdout(io.StringIO()): + verify_traffic_revision(value, expected_revision=value['metadata']['name'], + expected_commit=commit, require_startup_health=True) + meta, status = value['metadata'], value['status'] + require(bool(meta.get('uid')) and type(meta.get('generation')) is int + and meta['generation'] > 0 and status.get('observedGeneration') == meta['generation']) + for kind in ('Ready', 'Active', 'ContainerReady', 'ContainerHealthy'): + rows = [row for row in status['conditions'] if row.get('type') == kind] + require(len(rows) == 1 and rows[0].get('status') == 'True') + + +class API: + """Existing client credentials stay in memory; no retries or diagnostic body.""" + def __init__(self, project: str, region: str, service: str): + self.base = ('https://' + region + '-run.googleapis.com/apis/serving.knative.dev/v1/' + + 'namespaces/' + project) + self.service_url = self.base + '/services/' + service + self.v2_url = ('https://run.googleapis.com/v2/projects/' + project + + '/locations/' + region + '/services/' + service) + + def __call__(self, kind: str, name: str = '', body: dict | None = None, + timeout: float = 30) -> dict: + try: + token = subprocess.run(['gcloud', 'auth', 'print-access-token'], + capture_output=True, text=True, check=True, timeout=min(10, timeout / 2)).stdout.strip() + require(bool(token)) + url = self.v2_url if kind == 'minimum' else ( + self.base + '/revisions/' + name if kind in {'revision', 'absent'} else self.service_url) + request = Request(url, data=json.dumps(body).encode() if body is not None else None, + method='PUT' if body is not None else 'GET', + headers={'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}) + # Token acquisition plus HTTP fits inside the caller's remaining time. + with urlopen(request, timeout=timeout / 2) as response: + result = json.load(response) + require(isinstance(result, dict)) + return {'absent': False} if kind == 'absent' else result + except HTTPError as error: + if kind == 'absent' and error.code == 404: + return {'absent': True} + raise RecoveryError('recovery cloud operation failed or outcome unknown') from None + except Exception: + raise RecoveryError('recovery cloud operation failed or outcome unknown') from None + + +class Recovery: + def __init__(self, card: dict, attempt: Path, *, api=None, + clock=time.monotonic, sleep=time.sleep, now=lambda: datetime.now(timezone.utc)): + self.card, self.attempt = card, attempt + self.clock, self.sleep, self.now = clock, sleep, now + self.project, self.region, self.service = (card[key] for key in ('project', 'region', 'service')) + for value in (self.project, self.region, self.service, card['replacement']): + require(isinstance(value, str) and re.fullmatch(r'[a-z0-9][a-z0-9-]*', value) is not None) + require(card['scope'] == 's3-only' and card['restore_warm_on_failure'] is True) + require(re.fullmatch(r'[0-9a-f]{40}', card['source_commit']) is not None) + # Missing digest stops. It never silently builds or substitutes an image. + require(re.fullmatch(r'sha256:[0-9a-f]{64}', card['image_digest']) is not None) + require(re.fullmatch(r'[a-z0-9][a-z0-9-]{0,30}', card['request_id']) is not None) + require(card['replacement'] == self.service + '-rec-' + card['request_id'] + and len(card['replacement']) <= 63) + self.expires = datetime.fromisoformat(card['expires_at']) + require(self.expires.tzinfo is not None and self.expires > self.now()) + require(card['protected_environment'] and card['authority']) + self.api = api or API(self.project, self.region, self.service) + self.warm = card['warm']['metadata']['name'] + require(self.warm.startswith(self.service + '-') and self.warm != card['replacement']) + self.expected = None + self.replacement = None + self.activated = False + self.minima = [] + + def valid(self): + require(self.now() < self.expires) + + def record(self, status: str): + # Fixed states only, never raw identities/configuration/conditions. + self.attempt.write_text(json.dumps({'status': status, 'minimum_readbacks': self.minima}) + '\n') + + def minimum(self): + value = self.api('minimum') + require(value.get('name') == 'projects/' + self.project + '/locations/' + self.region + + '/services/' + self.service) + require(value.get('uid') == self.card['service_snapshot']['metadata']['uid']) + require(value.get('scaling', {}).get('minInstanceCount', 0) == 0 + and value.get('template', {}).get('scaling', {}).get('minInstanceCount', 0) == 0) + warm = self.api('revision', self.warm) + require(fingerprint(warm) == fingerprint(self.card['warm'])) + rows = [row.get('status') for row in warm.get('status', {}).get('conditions', []) + if row.get('type') == 'MinInstancesProvisioned'] + self.minima.append({'service_minimum': 0, 'template_minimum': 0, + 'warm_revision_minimum': 1, + 'warm_minimum_one_provisioned': rows[0] if len(rows) == 1 else 'Unknown', + 'billing_total': 'NOT_QUERIED'}) + + def stable(self, distribution: list, timeout=30) -> dict: + value = self.api('service', timeout=timeout) + expected = self.expected + require(value['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) + require(value['spec'] == expected['spec'] and active(value) == sorted(distribution)) + for key in ('name', 'namespace', 'generation', 'labels', 'annotations'): + require(service_meta(value, key) == service_meta(expected, key)) + require(value['metadata'].get('resourceVersion')) + require(value.get('status', {}).get('observedGeneration') == value['metadata']['generation']) + return value + + def check_warm(self, *, require_health=True): + value = self.api('revision', self.warm) + require(fingerprint(value) == fingerprint(self.card['warm'])) + require(str(value['metadata'].get('annotations', {}).get(MINIMUM, '0')) == '1') + require(value['status']['imageDigest'].split('@')[-1] == self.card['image_digest']) + if require_health: + healthy(value, self.card['source_commit']) + + def check_replacement(self, timeout=30): + value = self.api('revision', self.card['replacement'], timeout=timeout) + require(value['metadata'].get('name') == self.card['replacement']) + require(str(value['metadata'].get('annotations', {}).get(MINIMUM, '0')) == '0') + template = self.expected['spec']['template'] + require(value['spec'] == template['spec']) + for key in ('labels', 'annotations'): + for name, item in template['metadata'].get(key, {}).items(): + require(value['metadata'].get(key, {}).get(name) == item) + require(value['status']['imageDigest'].split('@')[-1] == self.card['image_digest']) + require(value['metadata'].get('uid') and value['metadata'].get('generation')) + if self.replacement is not None: + require(fingerprint(value) == fingerprint(self.replacement)) + return value + + def put(self, current: dict, spec: dict): + # Cloud Run v1 resourceVersion is an optimistic-concurrency precondition. + # Keep every setting; change only the explicitly constructed template/traffic. + meta = {key: current['metadata'][key] for key in + ('name', 'namespace', 'resourceVersion', 'labels', 'annotations') + if key in current['metadata']} + require(meta.get('resourceVersion')) + meta['annotations'] = service_meta(current, 'annotations') + self.api('service', body={'apiVersion': 'serving.knative.dev/v1', 'kind': 'Service', + 'metadata': meta, 'spec': spec}) + + def settle(self, spec: dict, distribution: list): + deadline = self.clock() + 60 + while True: + value = self.api('service') + require(value['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) + require(value['spec'] == spec) + require(value['metadata']['generation'] == self.expected['metadata']['generation'] + 1) + for key in ('name', 'namespace', 'labels', 'annotations'): + require(service_meta(value, key) == service_meta(self.expected, key)) + if (active(value) == sorted(distribution) + and value['status'].get('observedGeneration') == value['metadata']['generation']): + self.expected = value + return + require(self.clock() < deadline) + self.sleep(1) + + def shift(self, distribution: list, *, deadline=None): + self.valid() + previous = active(self.expected) + current = self.stable(previous) + self.check_warm() + if deadline is not None: + require(self.clock() < deadline) + healthy(self.check_replacement(timeout=min(30, deadline - self.clock())), + self.card['source_commit']) + require(self.clock() < deadline) + self.valid() + spec = copy.deepcopy(current['spec']) + spec['traffic'] = [{'revisionName': name, 'percent': percent} for name, percent in distribution] + self.put(current, spec) + self.settle(spec, distribution) + + def rollback(self): + # Expiry stops forward work, but not the expressly approved failure restore. + # A failed/unknown write is never repeated; fresh readback chooses the one + # distinct compensation. Never overwrite another actor's configuration. + self.check_warm(require_health=False) + current = self.api('service') + require(current['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) + require(current['spec']['template'] == self.expected['spec']['template']) + for key in ('name', 'namespace', 'labels', 'annotations'): + require(service_meta(current, key) == service_meta(self.expected, key)) + left, right = copy.deepcopy(current['spec']), copy.deepcopy(self.expected['spec']) + left.pop('traffic', None) + right.pop('traffic', None) + require(left == right) + allowed = [[(self.warm, 100)], sorted([(self.warm, 99), (self.card['replacement'], 1)]), + [(self.card['replacement'], 100)]] + require(active(current) in allowed) + desired = [{'revisionName': self.warm, 'percent': 100}] + if active(current) == [(self.warm, 100)] and current['spec'].get('traffic') == desired: + self.expected = current + return + spec = copy.deepcopy(current['spec']) + spec['traffic'] = desired + self.expected = current + self.put(current, spec) + self.settle(spec, [(self.warm, 100)]) + + def execute(self): + # Exclusive creation persists across failure/restart, including reruns on + # the same host. The owner must never reuse the attempt/request identity. + with self.attempt.open('x'): + os.chmod(self.attempt, 0o600) + self.record('STARTED') + try: + self.valid() + before = self.api('service') + require(before == self.card['service_snapshot']) + require(before['metadata'].get('resourceVersion') and before['metadata'].get('uid')) + require(active(before) == [(self.warm, 100)]) + require(before['spec']['traffic'] == [{'revisionName': self.warm, 'percent': 100}]) + require(str(before['spec']['template']['metadata'].get('annotations', {}).get(MINIMUM, '0')) == '0') + self.expected = before + self.minimum() + self.check_warm() + self.sleep(5) + self.check_warm() + current = self.stable([(self.warm, 100)]) + self.valid() + spec = copy.deepcopy(current['spec']) + template = spec['template'] + template['metadata']['name'] = self.card['replacement'] + template['metadata'].setdefault('annotations', {})[MINIMUM] = '0' + containers = template['spec']['containers'] + require(len(containers) == 1) + image = containers[0]['image'].split('@')[0] + # Reject tags; only the existing repository plus retained digest. + require(':' not in image.rsplit('/', 1)[-1] and '/' in image) + containers[0]['image'] = image + '@' + self.card['image_digest'] + require(template['metadata'].get('labels', {}).get('commit-sha') == self.card['source_commit']) + require(self.api('absent', self.card['replacement']) == {'absent': True}) + self.valid() + # Persist intent before PUT; no restart can replay an ambiguous write. + self.record('CREATE_INTENT') + self.put(current, spec) + self.settle(spec, [(self.warm, 100)]) + require(self.expected['status'].get('latestCreatedRevisionName') == self.card['replacement']) + self.replacement = self.check_replacement() + self.record('ACTIVATE_INTENT') + self.activated = True + self.shift([(self.warm, 99), (self.card['replacement'], 1)]) + deadline = self.clock() + 300 + count = 0 + while True: + self.valid() + require(self.clock() < deadline) + self.stable([(self.warm, 99), (self.card['replacement'], 1)], + timeout=min(30, deadline - self.clock())) + require(self.clock() < deadline) + value = self.check_replacement(timeout=min(30, deadline - self.clock())) + try: + healthy(value, self.card['source_commit']) + except StartupHealthPending: + count = 0 + else: + count += 1 + require(self.clock() < deadline) + if count == 2: + break + self.sleep(min(5, deadline - self.clock())) + self.record('PROMOTE_INTENT') + self.shift([(self.card['replacement'], 100)], deadline=deadline) + healthy(self.check_replacement(), self.card['source_commit']) + self.minimum() + self.record('SUCCEEDED') + except BaseException: + self.record('FAILED') + if self.activated: + try: + self.rollback() + self.minimum() + self.record('FAILED_WARM_RESTORED') + except BaseException: + self.record('FAILED_RESTORE_UNCONFIRMED') + if not self.activated and self.expected is not None: + try: + self.minimum() + self.record('FAILED') + except Exception: + self.record('FAILED_MINIMUM_READBACK_UNAVAILABLE') + raise RecoveryError('recovery stopped; inspect private attempt state; do not replay') from None + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--card', type=Path, required=True) + parser.add_argument('--attempt', type=Path, required=True) + args = parser.parse_args() + os.umask(0o077) + def stop(*_): + raise RecoveryError('recovery interrupted') + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + try: + require(args.card.stat().st_mode & 0o077 == 0) + card = json.loads(args.card.read_text()) + head = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True, + text=True, check=True, timeout=10).stdout.strip() + require(head == card['operator_commit']) + dirty = subprocess.run(['git', 'status', '--porcelain', '--untracked-files=no'], + capture_output=True, text=True, check=True, timeout=10).stdout + require(not dirty) + require(os.environ.get('RECOVERY_PROTECTED_ENVIRONMENT') == card['protected_environment']) + Recovery(card, args.attempt).execute() + except Exception: + parser.exit(1, 'Recovery stopped. Read the private attempt state; never replay.\n') + print('Recovery succeeded; minimum-zero readback verified.') + + +if __name__ == '__main__': + main() diff --git a/tests/test_cloud_run_warm_recovery.py b/tests/test_cloud_run_warm_recovery.py new file mode 100644 index 00000000..a0eed232 --- /dev/null +++ b/tests/test_cloud_run_warm_recovery.py @@ -0,0 +1,257 @@ +"""Synthetic control-plane tests; no cloud, broker or business HTTP calls.""" +import copy +from datetime import datetime, timezone +import json +from unittest import mock + +import pytest + +from scripts.cloud_run_candidate_warmup import MINIMUM +from scripts.cloud_run_warm_recovery import API, Recovery, RecoveryError, healthy + +DIGEST = 'sha256:' + 'a' * 64 +COMMIT = 'b' * 40 + + +def revision(name, minimum): + return {'metadata': {'name': name, 'uid': name + '-uid', 'generation': 1, + 'labels': {'commit-sha': COMMIT}, 'annotations': {MINIMUM: str(minimum)}}, + 'spec': {'containers': [{'image': 'registry.invalid/unit/image@' + DIGEST, + 'startupProbe': {'httpGet': {'path': '/health', 'port': 8080}}, + 'env': [{'name': 'RUNTIME_TARGET_ENABLED', 'value': 'true'}, + {'name': 'RUNTIME_TARGET_JSON', 'value': '{}'}]}]}, + 'status': {'imageDigest': DIGEST, 'observedGeneration': 1, 'conditions': [ + {'type': kind, 'status': 'True'} for kind in ('Ready', 'Active', 'ContainerReady', 'ContainerHealthy')]}} + + +def card(): + warm = revision('unit-service-warm', 1) + baseline = {'metadata': {'name': 'unit-service', 'uid': 'service-uid', + 'namespace': 'unit-project', 'resourceVersion': '1', 'generation': 1, + 'labels': {}, 'annotations': {'run.googleapis.com/operation-id': 'old'}}, + 'spec': {'template': {'metadata': {'name': 'unit-service-old', + 'annotations': {MINIMUM: '0'}, 'labels': {'commit-sha': COMMIT}}, + 'spec': copy.deepcopy(warm['spec'])}, + 'traffic': [{'revisionName': 'unit-service-warm', 'percent': 100}]}, + 'status': {'observedGeneration': 1, 'latestCreatedRevisionName': 'unit-service-old', + 'traffic': [{'revisionName': 'unit-service-warm', 'percent': 100}]}} + return {'scope': 's3-only', 'project': 'unit-project', 'region': 'unit-region', + 'service': 'unit-service', 'warm': warm, 'service_snapshot': baseline, + 'source_commit': COMMIT, 'operator_commit': 'c' * 40, 'image_digest': DIGEST, + 'request_id': 'attempt1', 'replacement': 'unit-service-rec-attempt1', + 'authority': 'unit-authority', 'protected_environment': 'unit-protected', + 'expires_at': '2099-01-01T00:00:00+00:00', 'restore_warm_on_failure': True} + + +class Fake: + def __init__(self, approval, case='ok'): + self.card = approval + self.service = copy.deepcopy(approval['service_snapshot']) + self.revisions = {approval['warm']['metadata']['name']: copy.deepcopy(approval['warm'])} + self.writes, self.reads, self.sleeps = [], [], [] + self.case, self.seconds = case, 0 + + def clock(self): + return self.seconds + + def sleep(self, seconds): + self.seconds += seconds + self.sleeps.append(seconds) + + def __call__(self, kind, name='', body=None, timeout=30): + assert 0 < timeout <= 30 + self.reads.append((kind, name)) + if kind == 'absent': + return {'absent': name not in self.revisions} + if kind == 'minimum': + return {'name': 'projects/unit-project/locations/unit-region/services/unit-service', + 'uid': 'service-uid', 'scaling': {'minInstanceCount': 1 if self.case == 'min' else 0}} + if kind == 'revision': + value = copy.deepcopy(self.revisions[name]) + split = len(self.writes) >= 2 + if name == self.card['replacement'] and split: + if self.case in {'missing', 'timeout'}: + value['status']['conditions'] = [row for row in value['status']['conditions'] + if row['type'] != 'ContainerHealthy'] + if self.case in {'false', 'unknown', 'duplicate'}: + value['status']['conditions'].append({'type': 'ContainerHealthy', + 'status': 'False' if self.case == 'false' else 'Unknown'}) + if self.case == 'digest': value['status']['imageDigest'] = 'sha256:' + 'd' * 64 + if self.case == 'revision_uid': value['metadata']['uid'] = 'drift' + if self.case == 'stale': value['status']['observedGeneration'] = 0 + if self.case == 'interrupt': raise KeyboardInterrupt() + return value + if body is not None: + assert body['metadata']['resourceVersion'] == self.service['metadata']['resourceVersion'] + assert 'run.googleapis.com/operation-id' not in body['metadata']['annotations'] + self.writes.append(copy.deepcopy(body)) + index = len(self.writes) + if self.case == 'reject_split' and index == 2: raise RecoveryError('rejected') + if self.case == 'reject_promote' and index == 3: raise RecoveryError('rejected') + if self.case == 'restore_unknown' and index == 4: raise RecoveryError('unknown') + self.service['spec'] = copy.deepcopy(body['spec']) + self.service['metadata']['generation'] += 1 + self.service['metadata']['resourceVersion'] = str(index + 1) + self.service['metadata']['annotations']['run.googleapis.com/operation-id'] = str(index) + self.service['status']['observedGeneration'] = self.service['metadata']['generation'] + self.service['status']['traffic'] = copy.deepcopy(body['spec']['traffic']) + template = body['spec']['template'] + new = template['metadata']['name'] + self.service['status']['latestCreatedRevisionName'] = new + if new not in self.revisions: + value = revision(new, 0) + value['spec'] = copy.deepcopy(template['spec']) + self.revisions[new] = value + if self.case == 'unknown_split' and index == 2: raise RecoveryError('unknown') + if self.case in {'unknown_promote', 'restore_unknown'} and index == 3: + raise RecoveryError('unknown') + return copy.deepcopy(self.service) + value = copy.deepcopy(self.service) + if len(self.writes) >= 2: + if self.case == 'service_uid': value['metadata']['uid'] = 'drift' + if self.case == 'service_spec': value['spec']['template']['spec']['concurrency'] = 2 + if self.case == 'other_traffic': + value['status']['traffic'] = [{'revisionName': 'unit-other', 'percent': 100}] + return value + + +def run(tmp_path, case='ok', approval=None): + approval = approval or card() + cloud = Fake(approval, case) + obj = Recovery(approval, tmp_path / 'attempt', api=cloud, clock=cloud.clock, sleep=cloud.sleep) + return obj, cloud + + +@pytest.fixture(autouse=True) +def continuity(): + with mock.patch('scripts.runtime_target_continuity.verify_target_continuity'): + yield + + +def test_success_uses_retained_digest_and_min_zero_only(tmp_path): + obj, cloud = run(tmp_path) + obj.execute() + assert len(cloud.writes) == 3 + assert cloud.writes[0]['spec']['template']['metadata']['annotations'][MINIMUM] == '0' + assert cloud.writes[0]['spec']['template']['spec'] == card()['service_snapshot']['spec']['template']['spec'] + assert cloud.writes[1]['spec']['traffic'] == [ + {'revisionName': 'unit-service-warm', 'percent': 99}, + {'revisionName': 'unit-service-rec-attempt1', 'percent': 1}] + assert cloud.writes[2]['spec']['traffic'] == [{'revisionName': 'unit-service-rec-attempt1', 'percent': 100}] + assert cloud.sleeps == [5, 5] + assert cloud.reads.count(('minimum', '')) == 2 + assert json.loads(obj.attempt.read_text())['status'] == 'SUCCEEDED' + with pytest.raises(FileExistsError): obj.execute() + assert len(cloud.writes) == 3 + + +@pytest.mark.parametrize('case', ['missing', 'timeout', 'false', 'unknown', 'duplicate', 'digest', + 'revision_uid', 'stale', 'interrupt', 'reject_split', 'unknown_split', 'reject_promote', 'unknown_promote']) +def test_failed_forward_path_restores_warm_once_and_never_replays(tmp_path, case): + obj, cloud = run(tmp_path, case) + with pytest.raises(RecoveryError): obj.execute() + assert cloud.service['status']['traffic'] == [{'revisionName': 'unit-service-warm', 'percent': 100}] + status = json.loads(obj.attempt.read_text())['status'] + assert status == 'FAILED_WARM_RESTORED' + assert sum(write['spec']['traffic'][0].get('percent') == 99 for write in cloud.writes) == 1 + assert cloud.seconds <= 310 + with pytest.raises(FileExistsError): obj.execute() + + +@pytest.mark.parametrize('case', ['service_uid', 'service_spec', 'other_traffic', 'restore_unknown']) +def test_drift_or_failed_compensation_is_unconfirmed_and_never_blindly_retried(tmp_path, case): + obj, cloud = run(tmp_path, case) + with pytest.raises(RecoveryError): obj.execute() + assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_RESTORE_UNCONFIRMED' + assert len(cloud.writes) == (4 if case == 'restore_unknown' else 2) + + +@pytest.mark.parametrize('field,value', [('image_digest', ''), ('scope', 'peer1'), + ('source_commit', 'main'), ('restore_warm_on_failure', False), + ('expires_at', '2000-01-01T00:00:00+00:00'), ('replacement', 'unit-other')]) +def test_malformed_or_expired_card_cannot_write(tmp_path, field, value): + approval = card() + approval[field] = value + with pytest.raises((RecoveryError, ValueError)): + run(tmp_path, approval=approval) + + +def test_baseline_drift_stops_before_creation(tmp_path): + obj, cloud = run(tmp_path) + cloud.service['metadata']['resourceVersion'] = 'changed' + with pytest.raises(RecoveryError): obj.execute() + assert cloud.writes == [] + + +def test_service_minimum_one_is_not_revision_minimum_one_exception(tmp_path): + obj, cloud = run(tmp_path, 'min') + with pytest.raises(RecoveryError): obj.execute() + assert cloud.writes == [] + + +def test_false_duplicate_or_unknown_health_never_counts_as_healthy(): + value = card()['warm'] + value['status']['conditions'].append({'type': 'Ready', 'status': 'False'}) + with pytest.raises(ValueError): healthy(value, COMMIT) + + +def test_approval_expiry_after_activation_restores_without_forward_write(tmp_path): + obj, cloud = run(tmp_path) + obj.expires = datetime(2030, 1, 1, tzinfo=timezone.utc) + obj.now = lambda: datetime(2029 if len(cloud.writes) < 2 else 2031, 1, 1, tzinfo=timezone.utc) + with pytest.raises(RecoveryError): obj.execute() + assert len(cloud.writes) == 3 + assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_WARM_RESTORED' + + +def test_provider_failure_is_redacted_and_never_retried(): + api = API('unit-project', 'unit-region', 'unit-service') + with mock.patch('scripts.cloud_run_warm_recovery.subprocess.run', + side_effect=RuntimeError('private-token-and-host')) as run: + with pytest.raises(RecoveryError, match='cloud operation') as error: + api('service') + assert 'private' not in str(error.value) + assert run.call_count == 1 + + +def test_request_identity_cannot_adopt_an_existing_revision(tmp_path): + obj, cloud = run(tmp_path) + cloud.revisions[obj.card['replacement']] = revision(obj.card['replacement'], 0) + with pytest.raises(RecoveryError): obj.execute() + assert cloud.writes == [] + + +def test_baseline_warm_health_loss_stops_before_create(tmp_path): + obj, cloud = run(tmp_path) + cloud.revisions[obj.warm]['status']['conditions'] = [] + with pytest.raises(RecoveryError): obj.execute() + assert cloud.writes == [] + + +def test_post_promotion_failure_can_restore_temporarily_retired_warm(tmp_path): + obj, cloud = run(tmp_path, 'unknown_promote') + original = cloud.__call__ + def api(kind, name='', body=None, timeout=30): + value = original(kind, name, body, timeout) + if kind == 'revision' and name == obj.warm and len(cloud.writes) == 3: + value['status']['conditions'] = [] + return value + obj.api = api + with pytest.raises(RecoveryError): obj.execute() + assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_WARM_RESTORED' + + +def test_cli_checks_pin_and_environment_before_cloud(tmp_path): + import sys + from scripts.cloud_run_warm_recovery import main + path = tmp_path / 'card.json' + path.write_text(json.dumps(card())) + path.chmod(0o600) + with mock.patch.object(sys, 'argv', ['recovery', '--card', str(path), + '--attempt', str(tmp_path / 'result')]), \ + mock.patch('scripts.cloud_run_warm_recovery.subprocess.run', + return_value=mock.Mock(stdout='wrong-head')), \ + mock.patch('scripts.cloud_run_warm_recovery.API') as api: + with pytest.raises(SystemExit) as error: main() + assert error.value.code == 1 + api.assert_not_called() From 4e695844b3d97d7d2a790b8a04234bfff15719bd Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:01:22 +0800 Subject: [PATCH 2/5] fix: Bound recovery convergence and preserve warm spec Co-Authored-By: Codex --- docs/ibkr-runtime-rollout.md | 4 ++-- scripts/cloud_run_warm_recovery.py | 9 ++++++-- tests/test_cloud_run_warm_recovery.py | 31 +++++++++++++++++++++++++++ 3 files changed, 40 insertions(+), 4 deletions(-) diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index 37966b36..c7b3b03d 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -615,8 +615,8 @@ an earlier operation. Any later attempt needs a fresh owner card and readback. The baseline must use an explicit warm-100 desired traffic row, with no tags or latest-revision indirection. The existing template must already carry the approved application source label, a single application container, an untagged -image repository and minimum zero. Configuration and Secret references are -copied without synchronization; runtime target continuity remains checked by +image repository and minimum zero. The template spec must equal the verified warm spec. Configuration and Secret +references are copied without synchronization; runtime target continuity remains checked by the common verifier. No Secret value, IAM policy, Scheduler, execution switch or risk setting is changed. diff --git a/scripts/cloud_run_warm_recovery.py b/scripts/cloud_run_warm_recovery.py index 76d04e86..d2d877ed 100644 --- a/scripts/cloud_run_warm_recovery.py +++ b/scripts/cloud_run_warm_recovery.py @@ -142,7 +142,8 @@ def minimum(self): if row.get('type') == 'MinInstancesProvisioned'] self.minima.append({'service_minimum': 0, 'template_minimum': 0, 'warm_revision_minimum': 1, - 'warm_minimum_one_provisioned': rows[0] if len(rows) == 1 else 'Unknown', + 'warm_minimum_one_provisioned': (rows[0] if len(rows) == 1 + and rows[0] in {'True', 'False', 'Unknown'} else 'Unknown'), 'billing_total': 'NOT_QUERIED'}) def stable(self, distribution: list, timeout=30) -> dict: @@ -193,7 +194,10 @@ def put(self, current: dict, spec: dict): def settle(self, spec: dict, distribution: list): deadline = self.clock() + 60 while True: - value = self.api('service') + remaining = deadline - self.clock() + require(remaining > 0) + value = self.api('service', timeout=min(30, remaining)) + require(self.clock() < deadline) require(value['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) require(value['spec'] == spec) require(value['metadata']['generation'] == self.expected['metadata']['generation'] + 1) @@ -271,6 +275,7 @@ def execute(self): current = self.stable([(self.warm, 100)]) self.valid() spec = copy.deepcopy(current['spec']) + require(spec['template']['spec'] == self.card['warm']['spec']) template = spec['template'] template['metadata']['name'] = self.card['replacement'] template['metadata'].setdefault('annotations', {})[MINIMUM] = '0' diff --git a/tests/test_cloud_run_warm_recovery.py b/tests/test_cloud_run_warm_recovery.py index a0eed232..d5c0cfc2 100644 --- a/tests/test_cloud_run_warm_recovery.py +++ b/tests/test_cloud_run_warm_recovery.py @@ -255,3 +255,34 @@ def test_cli_checks_pin_and_environment_before_cloud(tmp_path): with pytest.raises(SystemExit) as error: main() assert error.value.code == 1 api.assert_not_called() + + +def test_template_must_preserve_the_verified_warm_configuration(tmp_path): + approval = card() + approval['service_snapshot']['spec']['template']['spec']['containerConcurrency'] = 2 + obj, cloud = run(tmp_path, approval=approval) + with pytest.raises(RecoveryError): obj.execute() + assert cloud.writes == [] + + +def test_control_plane_convergence_never_accepts_a_late_read(tmp_path): + obj, cloud = run(tmp_path) + original = cloud.__call__ + def api(kind, name='', body=None, timeout=30): + value = original(kind, name, body, timeout) + if kind == 'service' and body is None and len(cloud.writes) == 1: + cloud.seconds += 61 + return value + obj.api = api + with pytest.raises(RecoveryError): obj.execute() + assert len(cloud.writes) == 1 + assert cloud.service['status']['traffic'] == [{'revisionName': 'unit-service-warm', 'percent': 100}] + + +def test_unexpected_provider_provisioning_value_is_not_published(tmp_path): + obj, cloud = run(tmp_path) + cloud.revisions[obj.warm]['status']['conditions'].append( + {'type': 'MinInstancesProvisioned', 'status': 'private-provider-value'}) + obj.execute() + assert 'private-provider-value' not in obj.attempt.read_text() + assert all(row['warm_minimum_one_provisioned'] == 'Unknown' for row in obj.minima) From bd516230a23e40c7b207740af060e484d06436df Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:07:48 +0800 Subject: [PATCH 3/5] fix: Protect recovery dispatch and failure compensation Co-Authored-By: Codex --- .github/workflows/recover-cloud-run-warm.yml | 112 +++++++++++++++++++ docs/ibkr-runtime-rollout.md | 16 ++- scripts/cloud_run_warm_recovery.py | 17 ++- tests/test_cloud_run_warm_recovery.py | 74 ++++++++++++ 4 files changed, 210 insertions(+), 9 deletions(-) create mode 100644 .github/workflows/recover-cloud-run-warm.yml diff --git a/.github/workflows/recover-cloud-run-warm.yml b/.github/workflows/recover-cloud-run-warm.yml new file mode 100644 index 00000000..dc70ed82 --- /dev/null +++ b/.github/workflows/recover-cloud-run-warm.yml @@ -0,0 +1,112 @@ +name: Recover retained Cloud Run warm revision + +on: + workflow_dispatch: + inputs: + card_uri: + description: "Protected GCS recovery card approved for this exact operator commit" + type: string + required: true + card_sha256: + description: "Owner-approved SHA-256 of the private recovery card bytes" + type: string + required: true + +permissions: + contents: read + id-token: write + +concurrency: + group: ibkr-retained-warm-recovery + cancel-in-progress: false + +jobs: + recover: + # Never rerun an attempt; a later owner action needs a new card/request. + if: github.event_name == 'workflow_dispatch' && github.run_attempt == 1 + environment: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }} + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - name: Mask and retain private input locators + uses: actions/github-script@v8 + with: + script: | + const fs = require('fs'); + const path = require('path'); + const input = context.payload.inputs; + for (const value of Object.values(input)) core.setSecret(value); + if (!/^gs:\/\/[a-z0-9][a-z0-9._-]+\/[a-zA-Z0-9/_.-]+$/.test(input.card_uri) || + !/^[a-f0-9]{64}$/.test(input.card_sha256)) { + throw new Error('Invalid private recovery card locator or checksum'); + } + fs.writeFileSync(path.join(process.env.RUNNER_TEMP, 'recovery-input.json'), + JSON.stringify(input), {mode: 0o600, flag: 'wx'}); + - uses: actions/checkout@v6 + with: + ref: ${{ github.sha }} + persist-credentials: false + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + - uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ secrets.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ secrets.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + - uses: google-github-actions/setup-gcloud@v3 + - name: Fetch and verify the approved private card + env: + RECOVERY_PROTECTED_ENVIRONMENT: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }} + run: | + python - <<'PY' + import hashlib + import json + import os + from pathlib import Path + import subprocess + os.umask(0o077) + try: + root = Path(os.environ['RUNNER_TEMP']) + value = json.loads((root / 'recovery-input.json').read_text()) + result = subprocess.run(['gcloud', 'storage', 'cat', value['card_uri']], + capture_output=True, check=True, timeout=30) + if hashlib.sha256(result.stdout).hexdigest() != value['card_sha256']: + raise ValueError + card = json.loads(result.stdout) + if (card['operator_commit'] != os.environ['GITHUB_SHA'] + or not os.environ.get('RECOVERY_PROTECTED_ENVIRONMENT') + or card['protected_environment'] != os.environ['RECOVERY_PROTECTED_ENVIRONMENT']): + raise ValueError + with (root / 'recovery-card.json').open('xb') as file: + file.write(result.stdout) + except Exception: + raise SystemExit('Private recovery card fetch or binding failed') from None + PY + - name: Install frozen operator dependencies + run: | + python -m pip install uv + uv sync --frozen --no-dev + - name: Execute the one-shot approved recovery + env: + RECOVERY_PROTECTED_ENVIRONMENT: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }} + run: | + uv run --no-sync python -m scripts.cloud_run_warm_recovery \ + --card "${RUNNER_TEMP}/recovery-card.json" \ + --attempt "${RUNNER_TEMP}/recovery-attempt.json" + - name: Report sanitized attempt state + if: always() + run: | + python - <<'PY' + import json + import os + from pathlib import Path + path = Path(os.environ['RUNNER_TEMP']) / 'recovery-attempt.json' + if path.exists(): + value = json.loads(path.read_text()) + allowed = {'STARTED', 'CREATE_INTENT', 'ACTIVATE_INTENT', 'PROMOTE_INTENT', + 'SUCCEEDED', 'FAILED', 'FAILED_WARM_RESTORED', + 'FAILED_RESTORE_UNCONFIRMED', 'FAILED_MINIMUM_READBACK_UNAVAILABLE'} + print('Recovery state: ' + (value['status'] if value.get('status') in allowed else 'UNKNOWN')) + else: + print('Recovery state: NOT_STARTED') + PY diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index c7b3b03d..c4b30fac 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -583,8 +583,14 @@ Run the operator code at the newly approved recovery merge, in the existing protected execution environment after that environment's approval. Source merge, `agent-approved`, a card file, and the `RECOVERY_PROTECTED_ENVIRONMENT` variable are not operational approval. The variable binds the runner's environment name; -it does not implement or bypass the hosting platform's environment gate. No new -workflow dispatch is introduced. A follow-up owner-approved operator job invokes: +it does not implement or bypass the hosting platform's environment gate. The +manual-only `recover-cloud-run-warm.yml` workflow uses the existing protected +environment and existing workload identity. It accepts only a protected GCS card +locator and its owner-approved SHA-256 checksum, masks both before use, downloads +the card privately and verifies its exact bytes/operator/environment before +execution. Dispatch at the approved operator merge; do not dispatch the ordinary +build workflow. Native environment approval remains required, and workflow reruns +are refused. The protected job invokes: ```bash RECOVERY_PROTECTED_ENVIRONMENT='' \ @@ -646,8 +652,10 @@ baseline. It need not remain Active while temporarily unreferenced. Only the original warm-100, exact 99/1, or exact replacement-100 distributions are eligible. Concurrent identity/configuration changes or unknown third-party traffic stop compensation rather than overwrite another actor. A rejected, -interrupted or ambiguous compensation is never retried. Inspect the private -attempt status; `FAILED_RESTORE_UNCONFIRMED` must not be reported as restored. +interrupted or ambiguous compensation is never retried. Failure-state persistence is best-effort: a full or unwritable private volume +cannot suppress compensation, and the existing attempt claim is retained. A stale +or unreadable attempt status requires protected readback and never permits replay. +Inspect the private attempt status; `FAILED_RESTORE_UNCONFIRMED` must not be reported as restored. A failure before activation never advances traffic or creates another revision. Before and after execution, including a readable failure state, independent diff --git a/scripts/cloud_run_warm_recovery.py b/scripts/cloud_run_warm_recovery.py index d2d877ed..ef1441f1 100644 --- a/scripts/cloud_run_warm_recovery.py +++ b/scripts/cloud_run_warm_recovery.py @@ -129,6 +129,12 @@ def record(self, status: str): # Fixed states only, never raw identities/configuration/conditions. self.attempt.write_text(json.dumps({'status': status, 'minimum_readbacks': self.minima}) + '\n') + def failure_record(self, status: str): + # Keep the existing exclusive attempt claim even when its volume fails. + # Persistence must never suppress the independently authorized restore. + with contextlib.suppress(BaseException): + self.record(status) + def minimum(self): value = self.api('minimum') require(value.get('name') == 'projects/' + self.project + '/locations/' + self.region @@ -322,20 +328,21 @@ def execute(self): self.minimum() self.record('SUCCEEDED') except BaseException: - self.record('FAILED') + self.failure_record('FAILED') if self.activated: try: self.rollback() self.minimum() - self.record('FAILED_WARM_RESTORED') + outcome = 'FAILED_WARM_RESTORED' except BaseException: - self.record('FAILED_RESTORE_UNCONFIRMED') + outcome = 'FAILED_RESTORE_UNCONFIRMED' + self.failure_record(outcome) if not self.activated and self.expected is not None: try: self.minimum() - self.record('FAILED') + self.failure_record('FAILED') except Exception: - self.record('FAILED_MINIMUM_READBACK_UNAVAILABLE') + self.failure_record('FAILED_MINIMUM_READBACK_UNAVAILABLE') raise RecoveryError('recovery stopped; inspect private attempt state; do not replay') from None diff --git a/tests/test_cloud_run_warm_recovery.py b/tests/test_cloud_run_warm_recovery.py index d5c0cfc2..03a1eb9a 100644 --- a/tests/test_cloud_run_warm_recovery.py +++ b/tests/test_cloud_run_warm_recovery.py @@ -286,3 +286,77 @@ def test_unexpected_provider_provisioning_value_is_not_published(tmp_path): obj.execute() assert 'private-provider-value' not in obj.attempt.read_text() assert all(row['warm_minimum_one_provisioned'] == 'Unknown' for row in obj.minima) + + +def fetch_card(tmp_path, monkeypatch, *, checksum=None, card_value=None, operator=None, environment=None): + import hashlib + from pathlib import Path + import textwrap + script = Path('.github/workflows/recover-cloud-run-warm.yml').read_text() + block = script.split(' - name: Fetch and verify the approved private card', 1)[1].split(' - name:', 1)[0] + python = textwrap.dedent(block.split(" python - <<'PY'\n", 1)[1].rsplit(' PY', 1)[0]) + payload = json.dumps(card_value or card()).encode() + (tmp_path / 'recovery-input.json').write_text(json.dumps({'card_uri': 'gs://unit-private/card.json', + 'card_sha256': checksum or hashlib.sha256(payload).hexdigest()})) + monkeypatch.setenv('RUNNER_TEMP', str(tmp_path)) + monkeypatch.setenv('GITHUB_SHA', operator or card()['operator_commit']) + monkeypatch.setenv('RECOVERY_PROTECTED_ENVIRONMENT', environment or card()['protected_environment']) + previous = __import__('os').umask(0o077) + try: + with mock.patch('subprocess.run', return_value=mock.Mock(stdout=payload)) as cloud: + exec(compile(python, '', 'exec'), {}) + return cloud, payload + finally: + __import__('os').umask(previous) + + +def test_protected_workflow_fetches_exact_card_without_logging(tmp_path, monkeypatch, capsys): + cloud, payload = fetch_card(tmp_path, monkeypatch) + result = tmp_path / 'recovery-card.json' + assert result.read_bytes() == payload + assert result.stat().st_mode & 0o077 == 0 + cloud.assert_called_once_with(['gcloud', 'storage', 'cat', 'gs://unit-private/card.json'], + capture_output=True, check=True, timeout=30) + assert capsys.readouterr() == ('', '') + + +@pytest.mark.parametrize('kwargs', [{'checksum': 'd' * 64}, {'operator': 'wrong'}, {'environment': 'wrong'}]) +def test_protected_fetch_rejects_wrong_bytes_operator_or_environment(tmp_path, monkeypatch, capsys, kwargs): + with pytest.raises(SystemExit, match='Private recovery card fetch or binding failed'): + fetch_card(tmp_path, monkeypatch, **kwargs) + assert not (tmp_path / 'recovery-card.json').exists() + assert capsys.readouterr() == ('', '') + + +def test_workflow_has_native_environment_and_no_build_or_automatic_retry(): + from pathlib import Path + workflow = Path('.github/workflows/recover-cloud-run-warm.yml').read_text() + assert "environment: ${{ vars.IBKR_DEPLOY_ENVIRONMENT }}" in workflow + assert "github.run_attempt == 1" in workflow + assert 'cancel-in-progress: false' in workflow + assert 'workflow_dispatch:' in workflow + assert all(trigger not in workflow for trigger in (' push:', ' schedule:', ' workflow_call:')) + assert all(command not in workflow for command in ('docker build', 'docker push', 'gcloud run deploy', + 'gcloud secrets', 'gcloud projects add-iam')) + assert workflow.index('core.setSecret(value)') < workflow.index('google-github-actions/auth') + assert 'actions/upload-artifact' not in workflow + + +@pytest.mark.parametrize('failure_at,case', [('FAILED', 'false'), ('PROMOTE_INTENT', 'ok'), ('SUCCEEDED', 'ok')]) +def test_state_write_failure_cannot_prevent_compensation(tmp_path, failure_at, case): + obj, cloud = run(tmp_path, case) + record = obj.record + failed = False + def broken(status): + nonlocal failed + if status == failure_at: failed = True + if failed: raise OSError('private-volume-full') + record(status) + obj.record = broken + with pytest.raises(RecoveryError): obj.execute() + assert cloud.service['status']['traffic'] == [{'revisionName': obj.warm, 'percent': 100}] + assert len(cloud.writes) == (4 if failure_at == 'SUCCEEDED' else 3) + assert obj.attempt.exists() + writes = len(cloud.writes) + with pytest.raises(FileExistsError): obj.execute() + assert len(cloud.writes) == writes From 8cf607ae5846289cc2787c75bb034047609a368e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:10:51 +0800 Subject: [PATCH 4/5] fix: Reject unrelated generations during warm restore Co-Authored-By: Codex --- scripts/cloud_run_warm_recovery.py | 2 ++ tests/test_cloud_run_warm_recovery.py | 14 ++++++++++++++ 2 files changed, 16 insertions(+) diff --git a/scripts/cloud_run_warm_recovery.py b/scripts/cloud_run_warm_recovery.py index ef1441f1..20d06449 100644 --- a/scripts/cloud_run_warm_recovery.py +++ b/scripts/cloud_run_warm_recovery.py @@ -240,6 +240,8 @@ def rollback(self): current = self.api('service') require(current['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) require(current['spec']['template'] == self.expected['spec']['template']) + require(current['metadata']['generation'] in {self.expected['metadata']['generation'], + self.expected['metadata']['generation'] + 1}) for key in ('name', 'namespace', 'labels', 'annotations'): require(service_meta(current, key) == service_meta(self.expected, key)) left, right = copy.deepcopy(current['spec']), copy.deepcopy(self.expected['spec']) diff --git a/tests/test_cloud_run_warm_recovery.py b/tests/test_cloud_run_warm_recovery.py index 03a1eb9a..94f53365 100644 --- a/tests/test_cloud_run_warm_recovery.py +++ b/tests/test_cloud_run_warm_recovery.py @@ -360,3 +360,17 @@ def broken(status): writes = len(cloud.writes) with pytest.raises(FileExistsError): obj.execute() assert len(cloud.writes) == writes + + +def test_compensation_rejects_unrelated_generation_advances(tmp_path): + obj, cloud = run(tmp_path) + original = cloud.__call__ + def api(kind, name='', body=None, timeout=30): + value = original(kind, name, body, timeout) + if kind == 'service' and body is None and len(cloud.writes) >= 2: + value['metadata']['generation'] += 9 + return value + obj.api = api + with pytest.raises(RecoveryError): obj.execute() + assert len(cloud.writes) == 2 + assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_RESTORE_UNCONFIRMED' From 5c9366a71c68c53c9dfd1f0a6a663f6e9c32d094 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:18:31 +0800 Subject: [PATCH 5/5] fix: Fence ambiguous activation before confirming restore Co-Authored-By: Codex --- docs/ibkr-runtime-rollout.md | 8 ++++- scripts/cloud_run_warm_recovery.py | 40 ++++++++++++++++------- tests/test_cloud_run_warm_recovery.py | 47 +++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 13 deletions(-) diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index c4b30fac..be5af3fa 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -650,7 +650,13 @@ entry reads the actual state and attempts **one distinct warm=100 compensation** The warm's frozen identity/spec/digest must still match the already qualified baseline. It need not remain Active while temporarily unreferenced. Only the original warm-100, exact 99/1, or exact replacement-100 distributions are -eligible. Concurrent identity/configuration changes or unknown third-party +eligible. A confirmed write requires the exact confirmed generation and complete +desired spec, including traffic. A successor generation is admitted only for an +outstanding recovery write whose full desired spec matches its recorded intent. +An ambiguous activation cannot be declared restored from a preceding warm-100 +read: the one CAS compensation must advance resourceVersion and confirm warm-100, +fencing the old write precondition. Without that evidence restoration remains +unconfirmed, including a no-op compensation with no version advance. Concurrent identity/configuration changes or unknown third-party traffic stop compensation rather than overwrite another actor. A rejected, interrupted or ambiguous compensation is never retried. Failure-state persistence is best-effort: a full or unwritable private volume cannot suppress compensation, and the existing attempt claim is retained. A stale diff --git a/scripts/cloud_run_warm_recovery.py b/scripts/cloud_run_warm_recovery.py index 20d06449..fbb84fe5 100644 --- a/scripts/cloud_run_warm_recovery.py +++ b/scripts/cloud_run_warm_recovery.py @@ -121,6 +121,7 @@ def __init__(self, card: dict, attempt: Path, *, api=None, self.replacement = None self.activated = False self.minima = [] + self.pending = None def valid(self): require(self.now() < self.expires) @@ -194,10 +195,11 @@ def put(self, current: dict, spec: dict): if key in current['metadata']} require(meta.get('resourceVersion')) meta['annotations'] = service_meta(current, 'annotations') + self.pending = {'before': copy.deepcopy(current), 'spec': copy.deepcopy(spec)} self.api('service', body={'apiVersion': 'serving.knative.dev/v1', 'kind': 'Service', 'metadata': meta, 'spec': spec}) - def settle(self, spec: dict, distribution: list): + def settle(self, spec: dict, distribution: list, *, fence_version=None): deadline = self.clock() + 60 while True: remaining = deadline - self.clock() @@ -206,12 +208,16 @@ def settle(self, spec: dict, distribution: list): require(self.clock() < deadline) require(value['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) require(value['spec'] == spec) - require(value['metadata']['generation'] == self.expected['metadata']['generation'] + 1) + generation = self.expected['metadata']['generation'] + require(value['metadata']['generation'] in ( + {generation, generation + 1} if fence_version else {generation + 1})) for key in ('name', 'namespace', 'labels', 'annotations'): require(service_meta(value, key) == service_meta(self.expected, key)) if (active(value) == sorted(distribution) - and value['status'].get('observedGeneration') == value['metadata']['generation']): + and value['status'].get('observedGeneration') == value['metadata']['generation'] + and (fence_version is None or value['metadata']['resourceVersion'] != fence_version)): self.expected = value + self.pending = None return require(self.clock() < deadline) self.sleep(1) @@ -239,27 +245,37 @@ def rollback(self): self.check_warm(require_health=False) current = self.api('service') require(current['metadata']['uid'] == self.card['service_snapshot']['metadata']['uid']) - require(current['spec']['template'] == self.expected['spec']['template']) - require(current['metadata']['generation'] in {self.expected['metadata']['generation'], - self.expected['metadata']['generation'] + 1}) for key in ('name', 'namespace', 'labels', 'annotations'): require(service_meta(current, key) == service_meta(self.expected, key)) - left, right = copy.deepcopy(current['spec']), copy.deepcopy(self.expected['spec']) - left.pop('traffic', None) - right.pop('traffic', None) - require(left == right) + unresolved = self.pending + if unresolved is None: + require(current['metadata']['generation'] == self.expected['metadata']['generation'] + and current['spec'] == self.expected['spec']) + else: + generation = unresolved['before']['metadata']['generation'] + # A successor belongs to this attempt only when its ENTIRE desired + # spec matches the outstanding write, including desired traffic. + require((current['metadata']['generation'] == generation + and current['spec'] == unresolved['before']['spec']) + or (current['metadata']['generation'] == generation + 1 + and current['spec'] == unresolved['spec'])) allowed = [[(self.warm, 100)], sorted([(self.warm, 99), (self.card['replacement'], 1)]), [(self.card['replacement'], 100)]] require(active(current) in allowed) desired = [{'revisionName': self.warm, 'percent': 100}] - if active(current) == [(self.warm, 100)] and current['spec'].get('traffic') == desired: + if (unresolved is None and active(current) == [(self.warm, 100)] + and current['spec'].get('traffic') == desired): self.expected = current return spec = copy.deepcopy(current['spec']) spec['traffic'] = desired self.expected = current self.put(current, spec) - self.settle(spec, [(self.warm, 100)]) + # Even a preceding warm-100 observation cannot resolve a timed-out + # activation. One CAS compensation must advance resourceVersion, fencing + # the old precondition before restoration can be reported. No retry. + self.settle(spec, [(self.warm, 100)], + fence_version=current['metadata']['resourceVersion']) def execute(self): # Exclusive creation persists across failure/restart, including reruns on diff --git a/tests/test_cloud_run_warm_recovery.py b/tests/test_cloud_run_warm_recovery.py index 94f53365..d023bf64 100644 --- a/tests/test_cloud_run_warm_recovery.py +++ b/tests/test_cloud_run_warm_recovery.py @@ -374,3 +374,50 @@ def api(kind, name='', body=None, timeout=30): with pytest.raises(RecoveryError): obj.execute() assert len(cloud.writes) == 2 assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_RESTORE_UNCONFIRMED' + + +def test_compensation_does_not_overwrite_foreign_pending_traffic(tmp_path): + obj, cloud = run(tmp_path) + original = cloud.__call__ + def api(kind, name='', body=None, timeout=30): + value = original(kind, name, body, timeout) + if (kind == 'service' and body is None and len(cloud.writes) >= 2 + and obj.expected['spec']['traffic'][0]['percent'] == 99): + value['metadata']['generation'] += 1 + value['spec']['traffic'] = [{'revisionName': obj.warm, 'percent': 50}, + {'revisionName': obj.card['replacement'], 'percent': 50}] + return value + obj.api = api + with pytest.raises(RecoveryError): obj.execute() + assert len(cloud.writes) == 2 + assert json.loads(obj.attempt.read_text())['status'] == 'FAILED_RESTORE_UNCONFIRMED' + + +@pytest.mark.parametrize('fence_accepted', [True, False]) +def test_ambiguous_uncommitted_activation_is_fenced_before_restore_confirmation(tmp_path, fence_accepted): + obj, cloud = run(tmp_path) + original = cloud.__call__ + pending = None + def api(kind, name='', body=None, timeout=30): + nonlocal pending + if body is not None and len(cloud.writes) == 1: + pending = copy.deepcopy(body) + cloud.writes.append(copy.deepcopy(body)) + raise RecoveryError('activation outcome unknown') + if body is not None and len(cloud.writes) == 2 and not fence_accepted: + cloud.writes.append(copy.deepcopy(body)) + return copy.deepcopy(cloud.service) + if kind == 'minimum' and pending is not None: + if pending['metadata']['resourceVersion'] == cloud.service['metadata']['resourceVersion']: + cloud.service['spec'] = copy.deepcopy(pending['spec']) + cloud.service['status']['traffic'] = copy.deepcopy(pending['spec']['traffic']) + cloud.service['metadata']['generation'] += 1 + cloud.service['metadata']['resourceVersion'] = 'late-activation' + pending = None + return original(kind, name, body, timeout) + obj.api = api + with pytest.raises(RecoveryError): obj.execute() + assert cloud.service['status']['traffic'] == [{'revisionName': obj.warm, 'percent': 100}] + assert len(cloud.writes) == 3 + expected = 'FAILED_WARM_RESTORED' if fence_accepted else 'FAILED_RESTORE_UNCONFIRMED' + assert json.loads(obj.attempt.read_text())['status'] == expected