invoke-cloud-run.yml

name: Invoke Cloud Run

on:
  workflow_dispatch:
    inputs:
      path:
        description: "HTTP path to call"
        required: false
        default: "/dry-run"
        type: string
      allow_live_execution:
        description: "Allow calling /run live execution entrypoint"
        required: false
        default: false
        type: boolean
  workflow_call:
    inputs:
      path:
        description: "HTTP path to call"
        required: false
        default: "/account-facts-sync"
        type: string
      allow_live_execution:
        description: "Allow calling /run live execution entrypoint"
        required: false
        default: false
        type: boolean

permissions:
  contents: read

env:
  GCP_PROJECT_ID: firstradequant
  GCP_WORKLOAD_IDENTITY_PROVIDER: projects/1088907247379/locations/global/workloadIdentityPools/github-actions/providers/github-main
  GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: firstrade-platform-deploy@firstradequant.iam.gserviceaccount.com
  GCP_SCHEDULER_SERVICE_ACCOUNT: firstrade-platform-scheduler@firstradequant.iam.gserviceaccount.com

concurrency:
  group: ${{ github.workflow }}-${{ github.ref_name }}
  cancel-in-progress: false

jobs:
  invoke:
    name: Invoke Firstrade Cloud Run
    runs-on: ubuntu-latest
    timeout-minutes: 15
    permissions:
      contents: read
      id-token: write
    env:
      CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
      CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}
      CLOUD_SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION }}
      FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN_SECRET_NAME: ${{ vars.FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN_SECRET_NAME }}
    steps:
      - name: Validate inputs
        run: |
          set -euo pipefail

          if [ -z "${CLOUD_RUN_REGION:-}" ] || [ -z "${CLOUD_RUN_SERVICE:-}" ]; then
            echo "CLOUD_RUN_REGION and CLOUD_RUN_SERVICE are required repository variables." >&2
            exit 1
          fi

      - 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: Resolve service URL
        id: service
        run: |
          set -euo pipefail

          raw_path="${{ inputs.path }}"
          if [ -z "${raw_path}" ]; then
            raw_path="/dry-run"
          fi
          if [[ "${raw_path}" != /* ]]; then
            raw_path="/${raw_path}"
          fi
          if [ "${raw_path}" = "/run" ] && [ "${{ inputs.allow_live_execution }}" != "true" ]; then
            echo "Calling ${raw_path} can trigger the live execution entrypoint. Re-run with allow_live_execution=true if this is intentional." >&2
            exit 1
          fi

          service_json="$(
            gcloud run services describe "${CLOUD_RUN_SERVICE}" \
              --region "${CLOUD_RUN_REGION}" \
              --format=json
          )"
          service_url="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
          import json
          import os

          service = json.loads(os.environ["SERVICE_JSON"])
          print((service.get("status") or {}).get("url") or "")
          PY
          )"
          if [ -z "${service_url}" ]; then
            echo "Unable to resolve Cloud Run service URL." >&2
            exit 1
          fi

          service_ingress="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
          import json
          import os

          service = json.loads(os.environ["SERVICE_JSON"])
          annotations = (service.get("metadata") or {}).get("annotations") or {}
          print(annotations.get("run.googleapis.com/ingress-status") or annotations.get("run.googleapis.com/ingress") or "")
          PY
          )"
          latest_ready_revision="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
          import json
          import os

          service = json.loads(os.environ["SERVICE_JSON"])
          print((service.get("status") or {}).get("latestReadyRevisionName") or "")
          PY
          )"
          deployed_commit="$(SERVICE_JSON="${service_json}" python3 - <<'PY'
          import json
          import os

          service = json.loads(os.environ["SERVICE_JSON"])
          template = ((service.get("spec") or {}).get("template") or {}).get("metadata") or {}
          print((template.get("labels") or {}).get("commit-sha") or "")
          PY
          )"

          invoke_method="direct"
          scheduler_job=""
          scheduler_location=""
          scheduler_expected_path=""
          restore_probe_uri=""
          ensure_invoke_bridge_job() {
            local bridge_job="$1"
            local bridge_uri="$2"
            local bridge_location="$3"
            if gcloud scheduler jobs describe "${bridge_job}" \
              --project="${GCP_PROJECT_ID}" \
              --location="${bridge_location}" >/dev/null 2>&1; then
              current_uri="$(
                gcloud scheduler jobs describe "${bridge_job}" \
                  --project="${GCP_PROJECT_ID}" \
                  --location="${bridge_location}" \
                  --format='value(httpTarget.uri)' 2>/dev/null || true
              )"
              if [ -n "${current_uri}" ] && [ "${current_uri}" != "${bridge_uri}" ]; then
                echo "Updating invoke-bridge Cloud Scheduler job ${bridge_job} to ${bridge_uri}."
                gcloud scheduler jobs update http "${bridge_job}" \
                  --project="${GCP_PROJECT_ID}" \
                  --location="${bridge_location}" \
                  --uri="${bridge_uri}" \
                  --http-method=POST \
                  --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
                  --oidc-token-audience="${service_url}" \
                  --quiet
              fi
              return 0
            fi
            echo "Creating invoke-bridge Cloud Scheduler job ${bridge_job} -> ${bridge_uri}."
            gcloud scheduler jobs create http "${bridge_job}" \
              --project="${GCP_PROJECT_ID}" \
              --location="${bridge_location}" \
              --uri="${bridge_uri}" \
              --http-method=POST \
              --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
              --oidc-token-audience="${service_url}" \
              --schedule="0 0 1 1 *" \
              --time-zone="UTC" \
              --attempt-deadline=600s \
              --quiet
          }
          ensure_afs_scheduler_job() {
            local afs_job="$1"
            local afs_uri="$2"
            local afs_location="$3"
            local secret_name="${FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN_SECRET_NAME:-}"
            if [ -z "${secret_name}" ]; then
              echo "FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN_SECRET_NAME is required for /account-facts-sync." >&2
              exit 1
            fi
            # App auth uses Authorization Bearer sync token; Cloud Scheduler then places the
            # OIDC identity token in X-Serverless-Authorization for Cloud Run IAM.
            local sync_token
            # Deploy SA has secretAccessor on firstrade-account-facts-sync (SiYi-granted).
            sync_token="$(
              gcloud secrets versions access latest \
                --project="${GCP_PROJECT_ID}" \
                --secret="${secret_name}" | tr -d '\r\n'
            )"
            sync_token="${sync_token#Bearer }"
            if [ -z "${sync_token}" ]; then
              echo "Secret ${secret_name} returned an empty sync token." >&2
              exit 1
            fi
            echo "sync_token_len=${#sync_token}"
            SYNC_TOKEN_SECRET_NAME="${secret_name}" python3 - <<'PY'
          import json, os, subprocess, sys
          svc = os.environ["CLOUD_RUN_SERVICE"]
          region = os.environ["CLOUD_RUN_REGION"]
          project = os.environ["GCP_PROJECT_ID"]
          want = os.environ["SYNC_TOKEN_SECRET_NAME"]
          svc_j = json.loads(subprocess.check_output([
              "gcloud", "run", "services", "describe", svc,
              "--project", project, "--region", region, "--format=json",
          ], text=True))
          rev = (svc_j.get("status") or {}).get("latestReadyRevisionName") or ""
          print(f"serving_revision={rev}")
          data = json.loads(subprocess.check_output([
              "gcloud", "run", "revisions", "describe", rev,
              "--project", project, "--region", region, "--format=json",
          ], text=True))
          envs = (((data.get("spec") or {}).get("containers") or [{}])[0].get("env") or [])
          row = next((e for e in envs if e.get("name") == "FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN"), None)
          if row is None:
              print("sync_token_mount=MISSING", file=sys.stderr)
              sys.exit(1)
          vf = row.get("valueFrom") or {}
          vs = row.get("valueSource") or {}
          skr = vf.get("secretKeyRef") or vs.get("secretKeyRef") or {}
          ref = skr.get("name") or skr.get("secret") or ""
          print(f"sync_token_env_shape={sorted(row.keys())}")
          print(f"sync_token_secret_ref={ref or '<none>'}")
          print(f"sync_token_literal_set={bool(row.get('value'))}")
          if ref and ref != want:
              print(f"secret_ref_mismatch want={want} got={ref}", file=sys.stderr)
              sys.exit(1)
          if not ref and not row.get("value"):
              print("serving revision has neither secret ref nor literal SYNC_TOKEN", file=sys.stderr)
              sys.exit(1)
          PY

            if gcloud scheduler jobs describe "${afs_job}" \
              --project="${GCP_PROJECT_ID}" \
              --location="${afs_location}" >/dev/null 2>&1; then
              echo "Updating account-facts-sync scheduler ${afs_job} (OIDC + app Authorization header)."
              gcloud scheduler jobs update http "${afs_job}" \
                --project="${GCP_PROJECT_ID}" \
                --location="${afs_location}" \
                --uri="${afs_uri}" \
                --http-method=POST \
                --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
                --oidc-token-audience="${service_url}" \
                --attempt-deadline=600s \
                --quiet

              AFS_JOB_NAME="${afs_job}" AFS_LOCATION="${afs_location}" AFS_SYNC_TOKEN="${sync_token}" python3 - <<'PY'
          import json, os, subprocess, urllib.request
          project = os.environ["GCP_PROJECT_ID"]
          location = os.environ["AFS_LOCATION"]
          job = os.environ["AFS_JOB_NAME"]
          token = os.environ["AFS_SYNC_TOKEN"]
          name = f"projects/{project}/locations/{location}/jobs/{job}"
          access = subprocess.check_output(["gcloud", "auth", "print-access-token"], text=True).strip()
          url = f"https://cloudscheduler.googleapis.com/v1/{name}"
          payload = json.dumps({"httpTarget": {"headers": {"Authorization": f"Bearer {token}"}}}).encode()
          patch = urllib.request.Request(
              url + "?updateMask=httpTarget.headers",
              data=payload,
              method="PATCH",
              headers={"Authorization": f"Bearer {access}", "Content-Type": "application/json"},
          )
          with urllib.request.urlopen(patch, timeout=60) as resp:
              updated = json.load(resp)
          auth = ((updated.get("httpTarget") or {}).get("headers") or {}).get("Authorization") or ""
          print(f"scheduler_authorization_len={len(auth)}")
          print(f"scheduler_authorization_bearer_prefix={str(auth.startswith('Bearer ')).lower()}")
          if not auth.startswith("Bearer ") or len(auth) != len("Bearer ") + len(token):
              raise SystemExit("scheduler Authorization header length/prefix mismatch after PATCH")
          PY
            else
              echo "Creating account-facts-sync scheduler ${afs_job} (OIDC + app Authorization header)."
              gcloud scheduler jobs create http "${afs_job}" \
                --project="${GCP_PROJECT_ID}" \
                --location="${afs_location}" \
                --uri="${afs_uri}" \
                --http-method=POST \
                --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
                --oidc-token-audience="${service_url}" \
                --schedule="0 0 1 1 *" \
                --time-zone="UTC" \
                --attempt-deadline=600s \
                --quiet

              AFS_JOB_NAME="${afs_job}" AFS_LOCATION="${afs_location}" AFS_SYNC_TOKEN="${sync_token}" python3 - <<'PY'
          import json, os, subprocess, urllib.request
          project = os.environ["GCP_PROJECT_ID"]
          location = os.environ["AFS_LOCATION"]
          job = os.environ["AFS_JOB_NAME"]
          token = os.environ["AFS_SYNC_TOKEN"]
          name = f"projects/{project}/locations/{location}/jobs/{job}"
          access = subprocess.check_output(["gcloud", "auth", "print-access-token"], text=True).strip()
          url = f"https://cloudscheduler.googleapis.com/v1/{name}"
          payload = json.dumps({"httpTarget": {"headers": {"Authorization": f"Bearer {token}"}}}).encode()
          patch = urllib.request.Request(
              url + "?updateMask=httpTarget.headers",
              data=payload,
              method="PATCH",
              headers={"Authorization": f"Bearer {access}", "Content-Type": "application/json"},
          )
          with urllib.request.urlopen(patch, timeout=60) as resp:
              updated = json.load(resp)
          auth = ((updated.get("httpTarget") or {}).get("headers") or {}).get("Authorization") or ""
          print(f"scheduler_authorization_len={len(auth)}")
          print(f"scheduler_authorization_bearer_prefix={str(auth.startswith('Bearer ')).lower()}")
          if not auth.startswith("Bearer ") or len(auth) != len("Bearer ") + len(token):
              raise SystemExit("scheduler Authorization header length/prefix mismatch after PATCH")
          PY
            fi
            unset sync_token
            # Confirm Authorization header is present without printing its value.
            header_keys="$(
              gcloud scheduler jobs describe "${afs_job}" \
                --project="${GCP_PROJECT_ID}" \
                --location="${afs_location}" \
                --format=json | python3 -c 'import json,sys; job=json.load(sys.stdin); headers=((job.get("httpTarget") or {}).get("headers") or {}); keys=sorted(k for k in headers if k.lower() != "user-agent"); print(",".join(keys) if keys else "<none>")'
            )"
            echo "afs_scheduler_custom_header_keys=${header_keys}"
            case ",${header_keys}," in
              *,Authorization,*|*,authorization,*) ;;
              *)
                echo "account-facts-sync scheduler is missing the Authorization header after update." >&2
                exit 1
                ;;
            esac
          }
          if [ "${service_ingress}" = "internal" ]; then
            scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"
            case "${raw_path}" in
              /run)
                scheduler_job="${CLOUD_RUN_SERVICE}-scheduler"
                scheduler_expected_path="/run"
                ;;
              /probe)
                scheduler_job="${CLOUD_RUN_SERVICE}-probe-scheduler"
                scheduler_expected_path="/probe"
                ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/probe" "${scheduler_location}"
                ;;
              /dry-run)
                scheduler_job="${CLOUD_RUN_SERVICE}-precheck-scheduler"
                scheduler_expected_path="/dry-run"
                ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/dry-run" "${scheduler_location}"
                ;;
              /session-renew)
                # Manual-only forced read-only session renewal via a dedicated bridge job (same
                # placeholder-schedule pattern as the probe/precheck bridges).
                scheduler_job="${CLOUD_RUN_SERVICE}-session-renew-scheduler"
                scheduler_expected_path="/session-renew"
                ensure_invoke_bridge_job "${scheduler_job}" "${service_url}/session-renew" "${scheduler_location}"
                ;;
              /account-facts-sync)
                # Dedicated scheduler: OIDC for Cloud Run IAM (X-Serverless-Authorization) plus
                # FIRSTRADE_ACCOUNT_FACTS_SYNC_TOKEN in Authorization for app auth.
                scheduler_job="${CLOUD_RUN_SERVICE}-account-facts-sync-scheduler"
                scheduler_expected_path="/account-facts-sync"
                ensure_afs_scheduler_job "${scheduler_job}" "${service_url}/account-facts-sync" "${scheduler_location}"
                ;;
              *)
                echo "Cloud Run service ${CLOUD_RUN_SERVICE} has internal ingress, so GitHub-hosted runners cannot curl ${raw_path} directly." >&2
                echo "Use one of the scheduler-backed paths: /run, /probe, /dry-run, /account-facts-sync, /session-renew." >&2
                exit 1
                ;;
            esac

            scheduler_uri="$(
              gcloud scheduler jobs describe "${scheduler_job}" \
                --project="${GCP_PROJECT_ID}" \
                --location="${scheduler_location}" \
                --format='value(httpTarget.uri)' 2>/dev/null || true
            )"
            if [ -z "${scheduler_uri}" ]; then
              echo "Cloud Scheduler job ${scheduler_job} was not found in ${scheduler_location}." >&2
              exit 1
            fi
            scheduler_path="$(SCHEDULER_URI="${scheduler_uri}" python3 - <<'PY'
          import os
          from urllib.parse import urlparse

          def normalize(path: str) -> str:
              clean = (path or "/").rstrip("/")
              return clean or "/"

          print(normalize(urlparse(os.environ["SCHEDULER_URI"]).path))
          PY
            )"
            requested_path="$(RAW_PATH="${scheduler_expected_path}" python3 - <<'PY'
          import os

          clean = (os.environ["RAW_PATH"] or "/").rstrip("/")
          print(clean or "/")
          PY
            )"
            if [ "${scheduler_path}" != "${requested_path}" ]; then
              echo "Cloud Scheduler job ${scheduler_job} targets ${scheduler_uri}, not ${scheduler_expected_path}." >&2
              exit 1
            fi
            invoke_method="scheduler"
          fi

          echo "Cloud Run service: ${CLOUD_RUN_SERVICE}"
          echo "Cloud Run region: ${CLOUD_RUN_REGION}"
          echo "Cloud Run URL: ${service_url}"
          echo "Cloud Run ingress: ${service_ingress:-<empty>}"
          echo "Latest ready revision: ${latest_ready_revision:-<empty>}"
          echo "Deployed commit: ${deployed_commit:-<empty>}"
          echo "Invoke method: ${invoke_method}"
          if [ -n "${scheduler_job}" ]; then
            echo "Cloud Scheduler job: ${scheduler_job}"
            echo "Cloud Scheduler location: ${scheduler_location}"
          fi

          {
            echo "url=${service_url}"
            echo "path=${raw_path}"
            echo "invoke_method=${invoke_method}"
            echo "scheduler_job=${scheduler_job}"
            echo "scheduler_location=${scheduler_location}"
            echo "restore_probe_uri=${restore_probe_uri}"
          } >> "$GITHUB_OUTPUT"

      - name: Authenticate for service invocation
        if: steps.service.outputs.invoke_method == 'direct'
        id: invoke-auth
        uses: google-github-actions/auth@v3
        with:
          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
          token_format: id_token
          id_token_audience: ${{ steps.service.outputs.url }}
          id_token_include_email: true

      - name: Invoke service
        if: steps.service.outputs.invoke_method == 'direct'
        run: |
          set -euo pipefail

          curl --fail-with-body --show-error --silent \
            --request POST \
            --header "Authorization: Bearer ${{ steps.invoke-auth.outputs.id_token }}" \
            "${{ steps.service.outputs.url }}${{ steps.service.outputs.path }}"

      - name: Invoke internal service through Cloud Scheduler
        if: steps.service.outputs.invoke_method == 'scheduler'
        env:
          RESTORE_PROBE_URI: ${{ steps.service.outputs.restore_probe_uri }}
          SERVICE_URL: ${{ steps.service.outputs.url }}
        run: |
          set -euo pipefail

          started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
          scheduler_job="${{ steps.service.outputs.scheduler_job }}"
          scheduler_location="${{ steps.service.outputs.scheduler_location }}"

          restore_probe_scheduler() {
            if [ -z "${RESTORE_PROBE_URI:-}" ]; then
              return 0
            fi
            echo "Restoring probe scheduler URI to ${RESTORE_PROBE_URI}"
            gcloud scheduler jobs update http "${scheduler_job}" \
              --project="${GCP_PROJECT_ID}" \
              --location="${scheduler_location}" \
              --uri="${RESTORE_PROBE_URI}" \
              --http-method=POST \
              --oidc-service-account-email="${GCP_SCHEDULER_SERVICE_ACCOUNT}" \
              --oidc-token-audience="${SERVICE_URL}" \
              --quiet || true
          }
          trap restore_probe_scheduler EXIT

          echo "Triggering ${scheduler_job} at ${started_at}."
          gcloud scheduler jobs run "${scheduler_job}" \
            --location="${scheduler_location}" \
            --quiet

          deadline=$((SECONDS + 180))
          while true; do
            job_json="$(
              gcloud scheduler jobs describe "${scheduler_job}" \
                --location="${scheduler_location}" \
                --format=json
            )"
            attempt_seen="$(JOB_JSON="${job_json}" STARTED_AT="${started_at}" python3 - <<'PY'
          import datetime as dt
          import json
          import os

          def parse_timestamp(value: str) -> dt.datetime | None:
              if not value:
                  return None
              text = value.replace("Z", "+00:00")
              return dt.datetime.fromisoformat(text)

          job = json.loads(os.environ["JOB_JSON"])
          last_attempt = parse_timestamp(job.get("lastAttemptTime") or "")
          started_at = parse_timestamp(os.environ["STARTED_AT"])
          print("true" if last_attempt and started_at and last_attempt >= started_at else "false")
          PY
            )"
            status_code="$(JOB_JSON="${job_json}" python3 - <<'PY'
          import json
          import os

          status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
          print(status.get("code") or "")
          PY
            )"
            status_message="$(JOB_JSON="${job_json}" python3 - <<'PY'
          import json
          import os

          status = (json.loads(os.environ["JOB_JSON"]).get("status") or {})
          print(status.get("message") or "")
          PY
            )"
            last_attempt_time="$(JOB_JSON="${job_json}" python3 - <<'PY'
          import json
          import os

          print(json.loads(os.environ["JOB_JSON"]).get("lastAttemptTime") or "")
          PY
            )"

            if [ "${attempt_seen}" = "true" ]; then
              if [ "${{ steps.service.outputs.path }}" = "/account-facts-sync" ]; then
                filter="resource.type=\"cloud_run_revision\" AND resource.labels.service_name=\"${CLOUD_RUN_SERVICE}\" AND timestamp>=\"${started_at}\" AND (httpRequest.requestUrl:\"/account-facts-sync\" OR textPayload:\"account_facts\" OR textPayload:\"account_facts_sync\" OR textPayload:\"qrs_reject\" OR textPayload:\"reason=\" OR textPayload:\"exc_type=\" OR textPayload:\"Firstrade\" OR textPayload:\"unauthorized\" OR jsonPayload.message:\"account_facts\")"
                gcloud logging read "${filter}" --project="${GCP_PROJECT_ID}" --limit=40 --format=json > /tmp/afs-logs.json || true
                python3 - <<'PY'
          import json
          try:
              rows = json.load(open("/tmp/afs-logs.json"))
          except Exception as exc:
              print(f"log_read_error={type(exc).__name__}")
              rows = []
          print(f"afs_log_entries={len(rows)}")
          for row in rows:
              http = row.get("httpRequest") or {}
              msg = row.get("textPayload") or (row.get("jsonPayload") or {}).get("message") or ""
              print(row.get("timestamp"), "status=", http.get("status"), "url=", http.get("requestUrl"), "msg=", str(msg)[:300])
          PY
              fi
              if [ -n "${status_code}" ] && [ "${status_code}" != "0" ]; then
                echo "Cloud Scheduler job ${scheduler_job} failed with status ${status_code}: ${status_message}" >&2
                exit 1
              fi
              echo "Cloud Scheduler job ${scheduler_job} ran at ${last_attempt_time}."
              break
            fi

            if [ "${SECONDS}" -ge "${deadline}" ]; then
              echo "Timed out waiting for Cloud Scheduler job ${scheduler_job} to record a new attempt." >&2
              exit 1
            fi

            echo "Waiting for Cloud Scheduler job ${scheduler_job} attempt..."
            sleep 10
          done