fix: gate shared scheduler cleanup

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

.github/workflows/sync-cloud-run-env.yml

@@ -1177,19 +1177,6 @@ jobs:
             esac
           done
 
-          if [ "${DEPLOYMENT_LABEL:-}" = "SG" ]; then
-            monitor_job_name="longbridge-monitor-dispatcher-scheduler"
-            if gcloud scheduler jobs describe "${monitor_job_name}" \
-              --project="${GCP_PROJECT_ID}" \
-              --location="${scheduler_location}" >/dev/null 2>&1; then
-              echo "Deleting obsolete shared monitor dispatcher ${monitor_job_name}."
-              gcloud scheduler jobs delete "${monitor_job_name}" \
-                --project="${GCP_PROJECT_ID}" \
-                --location="${scheduler_location}" \
-                --quiet
-            fi
-          fi
-
       - name: Reconcile legacy Cloud Scheduler jobs
         if: steps.config.outputs.env_sync_enabled == 'true'
         env:
@@ -1270,3 +1257,109 @@ jobs:
               --async \
               --quiet
           done <<< "${old_digests}"
+
+  cleanup-shared-monitor:
+    name: Retire shared monitor dispatcher
+    needs: sync
+    if: >
+      needs.sync.result == 'success' &&
+      (
+        github.event_name == 'workflow_run' ||
+        (github.event_name == 'workflow_dispatch' && inputs.target == 'configured')
+      )
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    environment: longbridge-sg
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON }}
+      CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
+      CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}
+    steps:
+      - name: Authenticate to Google Cloud
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+
+      - name: Set up gcloud
+        uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.GCP_PROJECT_ID }}
+          version: ">= 416.0.0"
+
+      - name: Verify per-service scheduler replacements
+        id: replacements
+        run: |
+          set -euo pipefail
+
+          replacements_ready="true"
+          mapfile -t configured_targets < <(python3 - <<'PY'
+          import json
+          import os
+
+          payload = json.loads(os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "{}")
+          for target in payload.get("targets") or []:
+              if not isinstance(target, dict):
+                  continue
+              service_name = str(target.get("service_name") or target.get("service") or "").strip()
+              scheduler = target.get("scheduler") if isinstance(target.get("scheduler"), dict) else {}
+              location = str(
+                  target.get("scheduler_location")
+                  or scheduler.get("location")
+                  or target.get("region")
+                  or target.get("cloud_run_region")
+                  or ""
+              ).strip()
+              if service_name:
+                  print(f"{service_name}\t{location}")
+          PY
+          )
+
+          if [ "${#configured_targets[@]}" -eq 0 ]; then
+            echo "No configured LongBridge targets; keeping the shared monitor dispatcher." >&2
+            replacements_ready="false"
+          fi
+
+          for configured_target in "${configured_targets[@]}"; do
+            IFS=$'\t' read -r service_name scheduler_location <<< "${configured_target}"
+            if [ -z "${scheduler_location}" ]; then
+              echo "No scheduler location for ${service_name}; keeping the shared monitor dispatcher." >&2
+              replacements_ready="false"
+              continue
+            fi
+            replacement_jobs=("${service_name}-probe-scheduler" "${service_name}-precheck-scheduler")
+            for replacement_job in "${replacement_jobs[@]}"; do
+              if ! gcloud scheduler jobs describe "${replacement_job}" \
+                --project="${GCP_PROJECT_ID}" \
+                --location="${scheduler_location}" >/dev/null 2>&1; then
+                echo "Replacement ${replacement_job} is not ready in ${scheduler_location}." >&2
+                replacements_ready="false"
+              fi
+            done
+          done
+
+          echo "ready=${replacements_ready}" >> "$GITHUB_OUTPUT"
+
+      - name: Delete obsolete shared monitor dispatcher
+        if: steps.replacements.outputs.ready == 'true'
+        run: |
+          set -euo pipefail
+
+          scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"
+          if [ -z "${scheduler_location}" ]; then
+            echo "Shared monitor dispatcher cleanup requires a scheduler location." >&2
+            exit 1
+          fi
+          monitor_job_name="longbridge-monitor-dispatcher-scheduler"
+          if gcloud scheduler jobs describe "${monitor_job_name}" \
+            --project="${GCP_PROJECT_ID}" \
+            --location="${scheduler_location}" >/dev/null 2>&1; then
+            echo "Deleting obsolete shared monitor dispatcher ${monitor_job_name}."
+            gcloud scheduler jobs delete "${monitor_job_name}" \
+              --project="${GCP_PROJECT_ID}" \
+              --location="${scheduler_location}" \
+              --quiet
+          fi

tests/test_scheduler_workflow.py

@@ -6,7 +6,14 @@ def test_replacement_probe_and_precheck_exist_before_shared_dispatcher_cleanup()
 
     probe = workflow.index('probe_job_name="${CLOUD_RUN_SERVICE}-probe-scheduler"')
     precheck = workflow.index('precheck_job_name="${CLOUD_RUN_SERVICE}-precheck-scheduler"')
+    cleanup_job = workflow.index("cleanup-shared-monitor:")
     cleanup = workflow.index('gcloud scheduler jobs delete "${monitor_job_name}"')
 
-    assert probe < cleanup
-    assert precheck < cleanup
+    assert probe < cleanup_job < cleanup
+    assert precheck < cleanup_job < cleanup
+
+    cleanup_section = workflow[cleanup_job:]
+    assert "needs: sync" in cleanup_section
+    assert "environment: longbridge-sg" in cleanup_section
+    assert 'replacement_jobs=("${service_name}-probe-scheduler" "${service_name}-precheck-scheduler")' in cleanup_section
+    assert "if: steps.replacements.outputs.ready == 'true'" in cleanup_section