From add91335357ab76a541ff7beed99803fa5618f25 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 24 Aug 2026 03:19:50 +0800 Subject: [PATCH 1/2] fix: claim execution identity before broker submission Co-Authored-By: Codex --- application/rebalance_service.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 1b88cc6..b90e79c 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -293,6 +293,7 @@ def fetch_replanned_state(): execution_marker_key = _build_execution_marker_key(config=config, execution=execution) execution_state_store = getattr(config, "execution_state_store", None) execution_already_recorded = False + execution_claim_acquired = False if execution_marker_key and execution_state_store: try: execution_already_recorded = bool(execution_state_store.has_marker(execution_marker_key)) @@ -319,6 +320,24 @@ def fetch_replanned_state(): f"Marker: {execution_marker_key}\n{type(exc).__name__}: {exc}", ) + if ( + not execution_already_recorded + and execution_marker_key + and execution_state_store + and bool(getattr(config, "execution_dedup_enabled", False)) + and not bool(getattr(config, "dry_run_only", False)) + ): + try: + execution_claim_acquired = bool(execution_state_store.claim_marker( + execution_marker_key, + metadata={"platform": "longbridge", "strategy_profile": getattr(config, "strategy_profile", "")}, + )) + execution_already_recorded = not execution_claim_acquired + except Exception as exc: + raise RuntimeError( + f"LongBridge execution claim unavailable; refusing broker submission: {type(exc).__name__}" + ) from exc + if execution_already_recorded: message = _execution_already_recorded_message( config=config, From bba417b5fb253ad89e2b6e913fad8f548950532e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 24 Aug 2026 03:24:13 +0800 Subject: [PATCH 2/2] guard runtime release set and image identity Co-Authored-By: Codex --- scripts/reconcile_cloud_runtime.py | 31 +++++++++++++++++++++++---- tests/test_reconcile_cloud_runtime.py | 22 ++++++++++++++++++- 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 3af0447..dc6bbf4 100755 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -176,7 +176,7 @@ def _traffic_on_latest(service: Mapping[str, Any], latest_revision: str) -> bool return False -def _revision_commit(*, project: str, region: str, revision: str, dry_run: bool) -> str: +def _revision_identity(*, project: str, region: str, revision: str, dry_run: bool) -> tuple[str, str, str]: payload = _run( [ "gcloud", @@ -191,8 +191,15 @@ def _revision_commit(*, project: str, region: str, revision: str, dry_run: bool) json_output=True, dry_run=dry_run, ) - labels = payload.get("metadata", {}).get("labels") or {} - return str(labels.get("commit-sha") or "").strip() + metadata = payload.get("metadata", {}) if isinstance(payload, Mapping) else {} + labels = metadata.get("labels") or {} + template = payload.get("spec", {}).get("containers", []) if isinstance(payload, Mapping) else [] + image = str(template[0].get("image") or "").strip() if template and isinstance(template[0], Mapping) else "" + return ( + str(labels.get("commit-sha") or "").strip(), + str(labels.get("release-set") or labels.get("release_set") or "").strip(), + image, + ) def ensure_latest_traffic( @@ -201,6 +208,8 @@ def ensure_latest_traffic( region: str, targets: Sequence[RuntimeTarget], expected_commit: str, + expected_release_set: str = "", + expected_image_digest: str = "", dry_run: bool, ) -> None: for target in targets: @@ -226,7 +235,7 @@ def ensure_latest_traffic( if not latest: raise ReconcileError(f"Unable to resolve latest revision for {target.service_name}") if expected_commit: - actual_commit = _revision_commit( + actual_commit, actual_release_set, actual_image = _revision_identity( project=project, region=target_region, revision=latest, @@ -237,6 +246,16 @@ def ensure_latest_traffic( f"{target.service_name} latest revision {latest} commit {actual_commit!r} " f"does not match expected {expected_commit!r}" ) + if expected_release_set and actual_release_set != expected_release_set: + raise ReconcileError( + f"{target.service_name} latest revision {latest} release-set {actual_release_set!r} " + f"does not match expected {expected_release_set!r}" + ) + if expected_image_digest and actual_image != expected_image_digest: + raise ReconcileError( + f"{target.service_name} latest revision {latest} image digest {actual_image!r} " + f"does not match expected {expected_image_digest!r}" + ) if not _traffic_on_latest(service, latest): print(f"Updating {target.service_name} traffic to latest revision {latest}.") _run( @@ -392,6 +411,8 @@ def parse_args(argv: Sequence[str]) -> argparse.Namespace: parser.add_argument("--region", default=os.environ.get("CLOUD_RUN_REGION", "")) parser.add_argument("--scheduler-location", default=os.environ.get("CLOUD_SCHEDULER_LOCATION", "")) parser.add_argument("--expected-commit", default=os.environ.get("GITHUB_SHA", "")) + parser.add_argument("--expected-release-set", default=os.environ.get("EXPECTED_RELEASE_SET", "")) + parser.add_argument("--expected-image-digest", default=os.environ.get("EXPECTED_IMAGE_DIGEST", "")) parser.add_argument("--ensure-latest-traffic", action="store_true") parser.add_argument("--delete-legacy-schedulers", action="store_true") parser.add_argument("--dry-run", action="store_true") @@ -411,6 +432,8 @@ def main(argv: Sequence[str] | None = None) -> int: region=args.region, targets=targets, expected_commit=args.expected_commit, + expected_release_set=args.expected_release_set, + expected_image_digest=args.expected_image_digest, dry_run=args.dry_run, ) if args.delete_legacy_schedulers: diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index a2bd1bc..42867a0 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -74,8 +74,10 @@ def fake_run(args, *, json_output=False, dry_run=False): "metadata": { "labels": { "commit-sha": "abc123", + "release-set": "release-1", } - } + }, + "spec": {"containers": [{"image": "gcr.io/example/app@sha256:abc"}]}, } if args[1:4] == ["run", "services", "update-traffic"]: return {} @@ -87,6 +89,8 @@ def fake_run(args, *, json_output=False, dry_run=False): region="asia-east1", targets=[reconcile.RuntimeTarget(service_name="longbridge-quant-paper-service")], expected_commit="abc123", + expected_release_set="release-1", + expected_image_digest="gcr.io/example/app@sha256:abc", dry_run=False, ) @@ -98,6 +102,22 @@ def fake_run(args, *, json_output=False, dry_run=False): ) self.assertEqual(describe_calls, 2) + def test_ensure_latest_traffic_rejects_stale_release_set_even_when_revision_is_healthy(self) -> None: + def fake_run(args, *, json_output=False, dry_run=False): + if args[1:4] == ["run", "services", "describe"]: + return {"status": {"latestReadyRevisionName": "service-00002", "traffic": []}} + if args[1:4] == ["run", "revisions", "describe"]: + return {"metadata": {"labels": {"commit-sha": "abc123", "release-set": "old"}}, "spec": {"containers": [{"image": "img@sha256:x"}]}} + self.fail(f"unexpected command: {args!r}") + + with patch.object(reconcile, "_run", side_effect=fake_run): + with self.assertRaisesRegex(reconcile.ReconcileError, "release-set"): + reconcile.ensure_latest_traffic( + project="p", region="r", + targets=[reconcile.RuntimeTarget(service_name="service")], + expected_commit="abc123", expected_release_set="new", dry_run=False, + ) + def test_ensure_latest_traffic_requires_latest_ready_revision(self) -> None: def fake_run(args, *, json_output=False, dry_run=False): if args[1:4] == ["run", "services", "describe"]: