"""Derive the deploy-chunk budget of one CI workflow run. Usage: python -m cli.meta.ci.slots [--matrix] [--format json] [--field ] A *chunk* is one serial block of deploy jobs in the orchestrator. Two independent ceilings bound it, and the smaller one wins: * **Queue clock.** GitHub cancels a job that sat queued for ``INFINITO_CI_QUEUE_HOURS``. With ``INFINITO_CI_CONCURRENCY`` runners free and every job assumed to burn its full ``timeout-minutes``, a chunk drains in waves, and the last wave that still starts inside the queue window is ``floor(queue_hours / job_timeout)``. A chunk therefore holds at most ``concurrency * waves`` jobs. Assuming every job runs to its timeout is deliberately pessimistic -- no job can outlast it -- so the drain estimate can only overshoot, never undershoot. * **Run job cap.** GitHub caps a workflow run at ``INFINITO_CI_JOB_LIMIT`` matrix jobs, counted across the whole run. Every non-deploy job the orchestrator chain spawns eats into that budget, so this module statically and CONSERVATIVELY counts them: - every orchestrator job reserves its full job count -- reusable workflows are opened and summed, static matrices multiplied out, dynamic (``fromJson``) matrices estimated at ``_DYNAMIC_MATRIX_ESTIMATE``; - the chunk blocks themselves reserve only their static jobs (the discover step); their dynamic per-row matrices are exactly the slots being budgeted. The worst entry point (``entry-*.yml``) adds its own jobs to the same run before calling the orchestrator; that overhead is subtracted too. What is left is :func:`available`, the rows one sweep can deploy. It is spent by :func:`chunk_count` blocks of :func:`chunk_size` rows. Because the orchestrator cannot generate a variable number of jobs, the block count is also capped by ``INFINITO_CI_MAX_CHUNKS``, the number of chunk blocks the workflow YAML actually declares (:func:`chunk_blocks`); a sweep that would need more rows than those blocks hold leaves the rest to the next sweep's offset. """ from __future__ import annotations import argparse import json import math import sys from typing import TYPE_CHECKING from utils.cache.files import PROJECT_ROOT from utils.cache.yaml import load_yaml_any from utils.env.parser import env_setting if TYPE_CHECKING: from pathlib import Path _ORCHESTRATOR = ".github/workflows/call-orchestrator.yml" _DEPLOY_WORKFLOW = ".github/workflows/call-test-deploy.yml" _DEPLOY_JOB = "deploy" _CHUNK_JOB_PREFIX = "test-deploy-chunk-" _DYNAMIC_MATRIX_ESTIMATE = 5 def setting(key: str) -> int: """One integer CI setting, read through the shared env SPOT.""" return int(env_setting(key)) def _load_workflow(path: Path) -> dict: data = load_yaml_any(str(path), default_if_missing={}) or {} return data if isinstance(data, dict) else {} def _jobs(workflow: dict) -> dict: jobs = workflow.get("jobs") return jobs if isinstance(jobs, dict) else {} def _matrix_size(job: dict) -> int: matrix = (job.get("strategy") or {}).get("matrix") if matrix is None: return 1 if not isinstance(matrix, dict): return _DYNAMIC_MATRIX_ESTIMATE if any(isinstance(v, str) and "${{" in v for v in matrix.values()): return _DYNAMIC_MATRIX_ESTIMATE include = matrix.get("include") size = len(include) if isinstance(include, list) else 0 axes = [ len(values) for key, values in matrix.items() if key not in ("include", "exclude") and isinstance(values, list) ] if axes: size += math.prod(axes) return max(size, 1) def _is_dynamic_matrix(job: dict) -> bool: matrix = (job.get("strategy") or {}).get("matrix") if matrix is None: return False if not isinstance(matrix, dict): return True return any(isinstance(v, str) and "${{" in v for v in matrix.values()) def _job_slots(repo_root: Path, job: dict, *, count_dynamic: bool) -> int: uses = job.get("uses") if isinstance(uses, str) and uses.startswith("./"): path = repo_root / uses.removeprefix("./") if not path.is_file(): return 1 return sum( _job_slots(repo_root, nested, count_dynamic=count_dynamic) for nested in _jobs(_load_workflow(path)).values() ) if _is_dynamic_matrix(job) and not count_dynamic: return 0 return _matrix_size(job) def job_timeout_minutes(repo_root: Path | None = None) -> int: """The deploy job's own ``timeout-minutes``, read straight out of the workflow so the wave arithmetic cannot drift from the value CI enforces.""" root = repo_root or PROJECT_ROOT job = _jobs(_load_workflow(root / _DEPLOY_WORKFLOW)).get(_DEPLOY_JOB) or {} timeout = job.get("timeout-minutes") if not isinstance(timeout, int): raise SystemExit( f"{_DEPLOY_WORKFLOW}: job {_DEPLOY_JOB!r} declares no integer " f"'timeout-minutes'; the chunk size is derived from it." ) return timeout def waves(repo_root: Path | None = None) -> int: """Runner waves that still start inside the queue window. Floored at 1: a job timeout at or above the window means one wave, never zero.""" window = setting("INFINITO_CI_QUEUE_HOURS") * 60 return max(window // job_timeout_minutes(repo_root), 1) def chunk_size(repo_root: Path | None = None) -> int: """Rows one chunk may hold before its tail risks the queue cancel. ``INFINITO_CI_CHUNK_SIZE`` takes precedence when set; 0 derives the value from the runner slots a chunk may assume and the waves the queue window affords. """ override = setting("INFINITO_CI_CHUNK_SIZE") if override > 0: return override return setting("INFINITO_CI_CONCURRENCY") * waves(repo_root) def reserved_breakdown(repo_root: Path | None = None) -> list[tuple[str, int]]: """Per orchestrator job: the job count it reserves in the run budget. Chunk blocks contribute only their static jobs (the discover step); their dynamic per-row matrices are the budgeted slots themselves.""" root = repo_root or PROJECT_ROOT jobs = _jobs(_load_workflow(root / _ORCHESTRATOR)) return [ ( name, _job_slots(root, job, count_dynamic=not name.startswith(_CHUNK_JOB_PREFIX)), ) for name, job in jobs.items() ] def reserved_slots(repo_root: Path | None = None) -> int: return sum(count for _, count in reserved_breakdown(repo_root)) def entry_overhead(repo_root: Path | None = None) -> int: """Worst-case jobs an ``entry-*.yml`` adds to the run around its orchestrator call (the orchestrator's own jobs are already counted).""" root = repo_root or PROJECT_ROOT orchestrator_path = root / _ORCHESTRATOR orchestrator_file = orchestrator_path.name totals = [0] for path in sorted(orchestrator_path.parent.iterdir()): if not (path.name.startswith("entry-") and path.name.endswith(".yml")): continue totals.append( sum( _job_slots(root, job, count_dynamic=True) for job in _jobs(_load_workflow(path)).values() if orchestrator_file not in str(job.get("uses", "")) ) ) return max(totals) def available(repo_root: Path | None = None) -> int: """Deploy rows one run may spend across all its chunks.""" limit = setting("INFINITO_CI_JOB_LIMIT") return max(limit - reserved_slots(repo_root) - entry_overhead(repo_root), 1) def chunk_blocks() -> int: """Chunk blocks call-orchestrator.yml declares, the most a sweep may plan.""" return setting("INFINITO_CI_MAX_CHUNKS") def chunk_count(repo_root: Path | None = None) -> int: """Chunk blocks a sweep fills, capped by the blocks the YAML declares.""" needed = math.ceil(available(repo_root) / chunk_size(repo_root)) return max(min(needed, chunk_blocks()), 1) def rows_per_sweep(repo_root: Path | None = None) -> int: """Rows one sweep covers: the chunk blocks it fills, bounded by the run job cap. Rows beyond this roll into the next sweep via its offset.""" return min(chunk_count(repo_root) * chunk_size(repo_root), available(repo_root)) def budget(repo_root: Path | None = None) -> dict[str, int]: return { "reserved": reserved_slots(repo_root), "entry_overhead": entry_overhead(repo_root), "available": available(repo_root), "job_timeout_minutes": job_timeout_minutes(repo_root), "waves": waves(repo_root), "chunk_size": chunk_size(repo_root), "chunk_count": chunk_count(repo_root), "rows_per_sweep": rows_per_sweep(repo_root), } def render_matrix() -> str: """The budget as a CLI table: per-job reservations, then the totals and the chunk arithmetic they leave room for.""" breakdown = reserved_breakdown() rows = [ *breakdown, ("reserved", reserved_slots()), ("entry overhead (worst entry-*.yml)", entry_overhead()), ("job limit (INFINITO_CI_JOB_LIMIT)", setting("INFINITO_CI_JOB_LIMIT")), ("available", available()), ("concurrency (INFINITO_CI_CONCURRENCY)", setting("INFINITO_CI_CONCURRENCY")), ("queue window (INFINITO_CI_QUEUE_HOURS)", setting("INFINITO_CI_QUEUE_HOURS")), ("job timeout (minutes)", job_timeout_minutes()), ("waves", waves()), ("chunk size", chunk_size()), ("chunk blocks (INFINITO_CI_MAX_CHUNKS)", chunk_blocks()), ("chunks filled", chunk_count()), ("rows per sweep", rows_per_sweep()), ] width = max(len(name) for name, _ in rows) lines = [f"{'job':<{width}} jobs", f"{'-' * width} ----"] lines += [f"{name:<{width}} {count:>4}" for name, count in breakdown] lines.append(f"{'-' * width} ----") lines += [f"{name:<{width}} {count:>4}" for name, count in rows[len(breakdown) :]] return "\n".join(lines) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Derive the deploy-chunk budget of one CI workflow run." ) parser.add_argument("--field", choices=sorted(budget())) parser.add_argument( "--matrix", action="store_true", help="Print the per-job budget breakdown as a table.", ) parser.add_argument("--format", choices=("json",), dest="fmt") args = parser.parse_args(argv) if args.matrix: print(render_matrix()) return 0 values = budget() if args.field: print(values[args.field]) return 0 print(json.dumps(values) if args.fmt == "json" else json.dumps(values, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())