diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 04c549f..1788f2f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,6 +78,12 @@ jobs: runs-on: ubuntu-24.04 permissions: contents: write + # One at a time across every ref: the clock check claims and removes the + # repository's marks for clock-check.yml, which a second live job running + # beside it would see change under it. + concurrency: + group: live-check + cancel-in-progress: false # Holds the full retry ladder of a failing `gh` call (github.py's # GH_TIMEOUT_SECONDS and RETRY_DELAYS_SECONDS) for the check and again for # its cleanup, so an API outage ends in a failed step with its message @@ -109,10 +115,16 @@ jobs: env: GH_TOKEN: ${{ github.token }} RECORD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + EVENT_NAME: ${{ github.event_name }} run: | + # Only a pull request may skip the clock check: its default branch can + # predate the workflow the check runs clocks against. + set -- + if [ "$EVENT_NAME" != pull_request ]; then set -- --require-clock-check; fi python3 -m tests.live_check \ --repo "$GITHUB_REPOSITORY" \ --sha "$RECORD_SHA" \ --workflow-file "ci-check-$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT.yml" \ --workflow-ref "$GITHUB_WORKFLOW_REF" \ - --workflow-sha "$GITHUB_WORKFLOW_SHA" + --workflow-sha "$GITHUB_WORKFLOW_SHA" \ + "$@" diff --git a/.github/workflows/clock-check.yml b/.github/workflows/clock-check.yml new file mode 100644 index 0000000..c799972 --- /dev/null +++ b/.github/workflows/clock-check.yml @@ -0,0 +1,20 @@ +name: Clock check + +# Enrolled in github-cron-trigger so CI's live check (tests/live_check.py) can +# run real clocks against this repository: the yearly line always has a newest +# slot, and the slot before it, for clocks to claim, and a dispatch of it runs +# the job below. It does nothing, so a delivery repeated by GitHub's yearly +# schedule changes nothing, and it needs no slot action. +on: + schedule: + - cron: "0 0 1 1 *" + repository_dispatch: + types: [github-cron-trigger/clock-check.yml] + +permissions: {} + +jobs: + delivered: + runs-on: ubuntu-24.04 + steps: + - run: echo "Delivered by a github-cron-trigger clock." diff --git a/README.md b/README.md index cc11fa3..6bab4f3 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,9 @@ A workflow whose repeated runs are harmless (a reconciler) needs only the clock. ref outside `refs/heads` and `refs/tags` is not a branch or a tag; nothing lists it in the UI or fetches it by default. `slot_ledger.py` says why a ref is the record and how the line is written into it. +- **Mark**: a clock's claim of a slot it sends, also a git ref, under + `refs/github-cron-trigger-clock/`. Marks are how several clocks share the + work without sending a slot twice ("The clock"). - **Part**: a run whose legs are recorded separately (one per environment, say) records `.` for each leg, so another delivery redoes only a leg that failed. @@ -219,50 +222,82 @@ line's newest slot whatever the parent was sent. ## The clock `tick` is the delivering half, run on a short interval by any scheduler outside -GitHub: +GitHub, on one host or several: ```sh -python3 -m github_cron_trigger tick --repo owner/name \ - --state ~/.local/state/github-cron-trigger/owner-name.json --send +python3 -m github_cron_trigger tick --repo owner/name --send ``` Each tick reads the workflow files of the repository's default branch through the API, in a single GraphQL query, so the host needs no clone. It works out -every `cron:` line's latest due slot in UTC and sends it to an enrolled +every enrolled `cron:` line's latest due slot in UTC and sends it to the workflow as a `repository_dispatch`: the newest due slot per line only, and never a slot the ledger already records. A merged change to a workflow reaches the clock at its next tick. -- **Shadow mode first.** Without `--send`, a tick only logs what it would send, - for every scheduled workflow, enrolled or not. Run it that way for a few days - and compare its log with GitHub's own `schedule` runs before turning sending - on. Removing `--send` later is the off switch. -- **Baselines.** A line is baselined without sending when the clock sees it for - the first time or its workflow's enrollment has changed, because its latest - slot may have run before anything could record it. -- **Sleep.** After a gap, such as a laptop asleep, the clock sends one delivery - per line, for the newest due slot, never a burst. A slot it slept through - entirely is left to GitHub's backstop: late, but still exactly once. -- **State.** The state file is a local cache. Losing it re-baselines every line, - which sends nothing. Use one state file per repository. -- **Exit status.** A tick that could not read a workflow, or could not send a - slot, logs the problem and exits 1, so the scheduler running it can report - it. A failed send is retried at the next tick. +### What it remembers, and where + +The clock keeps its memory in the repository it serves, not on its host. Before +it sends a slot it claims it, by creating a git ref: + +```text +refs/github-cron-trigger-clock/// +``` + +Creating a ref is atomic: when several clocks claim the same slot, exactly one +create succeeds, and only that clock sends. So: + +- **Run it on as many hosts as you like.** Each slot is still sent once. While + one host is asleep or down, another delivers on time. +- **A host keeps nothing.** One can join, leave or be rebuilt with nothing to + copy, and what has been sent can be read from the repository. +- **A failed send releases its claim**, so the next tick, on any host, sends + the slot again. A clock stopped between its claim and its send leaves that + slot to GitHub's backstop: late, but still delivered. + +These marks are not the ledger. The ledger records slots whose work finished, +and the slot action reads it; marks record slots a clock sent, for every +enrolled workflow, reconcilers included, and only clocks read them. A tick keeps +each line's two newest marks and removes the rest. + +### How it behaves + +- **Baselines.** A line no clock has marked is baselined - its latest slot is + marked without sending - because that slot may have run before anything could + record it. A workflow that stops being enrolled has its marks removed, so + enrolling it again baselines it again. +- **Sleep.** After a gap with no clock up, the next tick sends one delivery per + line, for the newest due slot, never a burst. Slots no clock was up for are + left to GitHub's backstop: late, but still exactly once. +- **Dry run first.** Without `--send`, a tick reads everything and reports what + it would do - baseline, send, or remove old marks - and writes nothing. + Removing `--send` later is the off switch. +- **What an idle tick costs.** Two small requests: the id of the default + branch's `.github/workflows` tree, which changes only when a workflow file + does, and the clock's marks. Workflow files are downloaded and parsed only + when they change, and then only the changed ones; everything read before is + reused from a cache under `$XDG_CACHE_HOME/github-cron-trigger` (or + `~/.cache/github-cron-trigger`), keyed by content id. The cache is only a + cache: deleting it costs one full read and changes nothing the clock decides. +- **Exit status.** A tick that could not read a workflow, the marks or the + ledger, or could not send a slot, logs the problem and exits 1, so the + scheduler running it can report it. ### Hosting it -Any host that stays up and can reach `api.github.com` will do. Install a -release, which puts `github-cron-trigger` on PATH: +Any host that can reach `api.github.com` will do, and more than one is better. +Install a release, which puts `github-cron-trigger` on PATH: ```sh -pip install "github-cron-trigger @ git+https://github.com/curlewlabs-com/github-cron-trigger@v0.1.0" +pip install "github-cron-trigger @ git+https://github.com/curlewlabs-com/github-cron-trigger@v0.2.0" ``` Or check out a release tag and run it as `PYTHONPATH=src python3 -m -github_cron_trigger`. Then run `tick` every few minutes. A crontab line: +github_cron_trigger`. Then run `tick` every few minutes, once per repository +served. A crontab line: ```crontab -*/5 * * * * github-cron-trigger tick --repo owner/name --state "$HOME/.local/state/github-cron-trigger/owner-name.json" --send >> "$HOME/github-cron-trigger.log" 2>&1 +*/5 * * * * github-cron-trigger tick --repo owner/name --send >> "$HOME/github-cron-trigger.log" 2>&1 ``` On macOS, a launchd agent with `StartInterval` 300 does the same. On Linux, a @@ -278,8 +313,9 @@ The clock calls the API through the `gh` CLI, so it uses whatever `gh` is logged in as, or `GH_TOKEN` when that is set. A fine-grained personal access token limited to the enrolled repositories needs: -- **Contents: Read and write.** Sending a `repository_dispatch` requires it. - Reading the workflow files and the ledger is the read half. +- **Contents: Read and write.** Sending a `repository_dispatch` requires it, + and so does creating and removing the clock's marks. Reading the workflow + files, the marks and the ledger is the read half. - **Metadata: Read**, which every fine-grained token carries. ## Missed slots @@ -377,8 +413,9 @@ an every-minute line runs is set by how often the clock ticks. releases page has a binary for every platform. The apt package named `yq` on Debian and Ubuntu is a different program, and the slot action refuses it by name. `MINIMUM_VERSION` in `yq.py` says why the floor sits where it does. -- **The clock's host**: `python3`, `gh` and yq as above, a scheduler, and a - token as described under "The clock's token". +- **The clock's hosts**: `python3`, `gh` and yq as above, a scheduler, and a + token as described under "The clock's token". Nothing else; a host keeps no + state. ## Versioning diff --git a/SECURITY.md b/SECURITY.md index d9d4ca9..7154734 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -36,5 +36,10 @@ security surface is: first checks that the version line is mikefarah's at the minimum version, so a different program called `yq` is never handed arguments. Workflow text reaches yq on standard input, never as a path taken from the event. -- **The clock's token.** The clock uses whatever `gh` is logged in as, or - `GH_TOKEN`. The README lists the narrowest fine-grained grant it needs. +- **The clock's token, and what it writes.** The clock uses whatever `gh` is + logged in as, or `GH_TOKEN`. The README lists the narrowest fine-grained + grant it needs. Besides dispatches, it writes only its marks: refs under + `refs/github-cron-trigger-clock/`, named from a workflow file name, a cron + line and a slot, each matched against a narrow pattern first. A mark's blob + names the slot and nothing about the host, since anyone can read a public + repository's refs. diff --git a/pyproject.toml b/pyproject.toml index 51d70c8..57a9ae5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi" [project] name = "github-cron-trigger" -version = "0.1.0" +version = "0.2.0" description = "Run each scheduled slot of a GitHub Actions workflow on time and exactly once, with GitHub's schedule as the backstop." readme = "README.md" license = "MIT" diff --git a/src/github_cron_trigger/clock.py b/src/github_cron_trigger/clock.py index 82e507a..02112f2 100644 --- a/src/github_cron_trigger/clock.py +++ b/src/github_cron_trigger/clock.py @@ -1,24 +1,29 @@ """The clock: deliver each workflow's due scheduled slots on time. Run on a short interval by a scheduler outside GitHub (launchd, a systemd -timer, any cron). Each tick reads the workflow files of the repository's -default branch through the API, so the host needs no clone of it; works out for -every `cron:` line the latest slot that has come due; and sends a -`repository_dispatch` of that slot to the workflow when it is enrolled and the -slot has not been delivered yet. GitHub's own `schedule` stays as the backstop; -the slot action and the ledger make every delivery of one slot do its work -once (README.md). +timer, any cron), on as many hosts as you like. Each tick reads the workflow +files of the repository's default branch through the API, so the host needs no +clone of it; works out for every enrolled `cron:` line the latest slot that has +come due; and sends a `repository_dispatch` of that slot to the workflow when no +clock has sent it yet. GitHub's own `schedule` stays as the backstop; the slot +action and the ledger make every delivery of one slot do its work once +(README.md). WHAT IT SENDS, AND WHEN. Per cron line, the newest due slot only: after a sleep the clock sends one delivery for a line, never a burst, and a slot it slept -through entirely is left to GitHub's backstop. A line is baselined - its current -latest slot is marked as handled without sending - when the clock sees it for -the first time, and when its workflow's enrollment has changed since the line -was last handled (the host slept across the merge that enrolled it, say). -Either way the slot may have come due while nothing could record it: the -workflow may already have run it, before the clock was watching or before it +through entirely is left to GitHub's backstop. A line with no mark at all +(clock_marks.py) is baselined - its current latest slot is marked without +sending - because that slot may have come due while nothing could record it: +the workflow may already have run it, before any clock served it or before it carried the slot action, and with no ledger record of that run a dispatch would -repeat finished work. +repeat finished work. A workflow that stops being enrolled has its marks +removed, so enrolling it again baselines it again (a host asleep across the +merge that enrolled it, say, sends nothing that predates the enrollment). + +WHO SENDS. Every decision is made against the marks in the repository, not +anything on the host, and a slot is sent only by the clock whose claim of it +succeeded. So two hosts ticking the same repository at once send each slot +once, and a host can join, leave or be rebuilt without carrying anything over. ENROLLED means the workflow declares `repository_dispatch` with the type `github-cron-trigger/`. An enrolled workflow that runs the @@ -26,37 +31,29 @@ `schedule` delivery of any other line, so GitHub's backstop run of such a workflow would fail every time, and the clock reports it rather than delivering it. One without the slot action - a reconciler, for which a repeated run is -harmless - may fire more often, since each dispatch names its exact slot. Every -other scheduled workflow is tracked too, and in shadow mode each line's due -slots are logged for all of them, which is how the slot math is checked against -GitHub's own schedule before anything is sent. - -STATE is a local JSON file holding, per line, the last slot handled and whether -the line's workflow was enrolled then. It only decides what to send next; -losing it re-baselines every line, which sends nothing, so it is a cache rather -than a record. +harmless - may fire more often, since each dispatch names its exact slot. + +A DRY RUN (no `--send`) reads everything a tick reads and reports what it would +do, and writes nothing: no mark, no dispatch, no clean-up. """ import argparse -import json -import os -import tempfile -from collections.abc import Mapping, Sequence +from collections.abc import Collection, Iterable, Mapping, Sequence from dataclasses import dataclass from datetime import datetime -from pathlib import Path -from typing import Literal +from typing import Literal, Protocol -from . import slot_ledger, workflow_files, yq +from . import default_branch, slot_ledger, yq +from .clock_marks import KEEP_PER_LINE, GitHubMarks, Mark, mark_for from .cron_slots import ( UTC, CronSchedule, UnsupportedCron, format_slot, latest_slot, - parse_slot, ) from .github import GitHubError, check_repo, gh_api +from .slot_ledger import cron_segment from .workflow_triggers import ( MisconfiguredWorkflow, dispatch_types, @@ -160,24 +157,61 @@ def scheduled_workflow( return ScheduledWorkflow(workflow_file, tuple(schedules), enrolled) -def state_key(workflow_file: str, schedule: CronSchedule) -> str: - return f"{workflow_file} {schedule.canonical}" +class Remote(Protocol): + """What a tick reads and writes outside the host: the repository's marks, + its slot ledger, and its dispatches. GitHubRemote is the real one.""" + def read_marks(self, repo: str) -> list[Mark]: ... -@dataclass(frozen=True) -class Handled: - """The last slot a tick handled for one line, and whether the line's - workflow was enrolled at the time.""" + def claim(self, repo: str, mark: Mark) -> bool: ... - slot: datetime - enrolled: bool + def release(self, repo: str, mark: Mark) -> None: ... + + def remove(self, repo: str, marks: Iterable[Mark]) -> None: ... + + def is_done( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> bool: ... + + def dispatch( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> None: ... + + +class GitHubRemote(GitHubMarks): + """The repository through the GitHub API.""" + + def read_marks(self, repo: str) -> list[Mark]: + return self.read(repo) + + def is_done( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> bool: + return slot_ledger.is_done(repo, workflow_file, schedule, slot) + + def dispatch( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> None: + """Send one slot to one workflow as a repository_dispatch.""" + gh_api( + [ + "--method", + "POST", + f"repos/{check_repo(repo)}/dispatches", + "-f", + f"event_type={EVENT_TYPE_PREFIX}{workflow_file}", + "-f", + f"client_payload[slot]={format_slot(slot)}", + "-f", + f"client_payload[cron]={schedule.canonical}", + ] + ) @dataclass(frozen=True) class Step: - """What a tick owes one cron line: `baseline` a line seen for the first - time or since its workflow's enrollment changed, or `deliver` its newest - slot.""" + """What a tick owes one enrolled cron line: `baseline` a line no clock has + marked, or `deliver` its newest slot.""" workflow: ScheduledWorkflow schedule: CronSchedule @@ -185,140 +219,147 @@ class Step: kind: Literal["baseline", "deliver"] +def newest_marks(marks: Iterable[Mark]) -> dict[tuple[str, str], datetime]: + """The newest marked slot of each line, keyed (workflow file, segment).""" + newest: dict[tuple[str, str], datetime] = {} + for mark in marks: + key = (mark.workflow_file, mark.segment) + if key not in newest or newest[key] < mark.slot: + newest[key] = mark.slot + return newest + + def plan( - workflows: list[ScheduledWorkflow], state: Mapping[str, Handled], now: datetime + workflows: Sequence[ScheduledWorkflow], marks: Iterable[Mark], now: datetime ) -> list[Step]: - """The lines whose newest slot this tick owes a step, in workflow order.""" + """The enrolled lines whose newest slot this tick owes a step, in workflow + order.""" + newest = newest_marks(marks) steps: list[Step] = [] for workflow in workflows: + if not workflow.enrolled: + continue for schedule in workflow.schedules: slot = latest_slot(schedule, now) - last = state.get(state_key(workflow.workflow_file, schedule)) - if last is None or last.enrolled != workflow.enrolled: + last = newest.get((workflow.workflow_file, cron_segment(schedule))) + if last is None: steps.append(Step(workflow, schedule, slot, "baseline")) - elif last.slot < slot: + elif last < slot: steps.append(Step(workflow, schedule, slot, "deliver")) return steps -def _read_handled(value: object) -> Handled: - if not isinstance(value, Mapping): - raise TypeError(f"entry {value!r} is not a map") - slot, enrolled = value.get("slot"), value.get("enrolled") - if not isinstance(slot, str) or not isinstance(enrolled, bool): - raise TypeError(f"entry {value!r} lacks a slot string or an enrolled flag") - return Handled(parse_slot(slot), enrolled) - - -def load_state(path: Path) -> dict[str, Handled]: - """The last slot handled per line; empty for a missing or unreadable file. +def stale_marks( + workflows: Sequence[ScheduledWorkflow], + marks: Iterable[Mark], + unjudged: Collection[str] = (), +) -> list[Mark]: + """The marks a tick removes: every mark of a line no enrolled workflow + carries any more - disenrolled, changed or deleted - and, of each line's + marks, all but the KEEP_PER_LINE newest. - An unreadable file is reported and treated as empty: every line then - re-baselines, which sends nothing, so the safe reading of a lost state is - no state. + A workflow file in `unjudged` - present, but unreadable or misconfigured + this tick - keeps every mark: removing them would make its lines baseline, + not deliver, once the file reads again. """ - try: - raw = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(raw, dict): - raise TypeError("not a JSON object") - return {key: _read_handled(value) for key, value in raw.items()} - except FileNotFoundError: - return {} - except (OSError, ValueError, TypeError) as exc: - print(f"clock: state file {path} unreadable ({exc}); re-baselining every line") - return {} - - -def save_state(path: Path, state: Mapping[str, Handled]) -> None: - """Write the state atomically, so a crash mid-write leaves the old file.""" - serialized = { - key: {"slot": format_slot(handled.slot), "enrolled": handled.enrolled} - for key, handled in sorted(state.items()) + current = { + (workflow.workflow_file, cron_segment(schedule)) + for workflow in workflows + if workflow.enrolled + for schedule in workflow.schedules } - path.parent.mkdir(parents=True, exist_ok=True) - fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".state-", suffix=".json") - try: - with os.fdopen(fd, "w", encoding="utf-8") as handle: - json.dump(serialized, handle, indent=2) - handle.write("\n") - os.replace(temporary, path) - except BaseException: - Path(temporary).unlink(missing_ok=True) - raise - - -def dispatch( - repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime -) -> None: - """Send one slot to one workflow as a repository_dispatch.""" - gh_api( - [ - "--method", - "POST", - f"repos/{check_repo(repo)}/dispatches", - "-f", - f"event_type={EVENT_TYPE_PREFIX}{workflow_file}", - "-f", - f"client_payload[slot]={format_slot(slot)}", - "-f", - f"client_payload[cron]={schedule.canonical}", - ] - ) + by_line: dict[tuple[str, str], list[Mark]] = {} + for mark in marks: + if mark.workflow_file in unjudged: + continue + by_line.setdefault((mark.workflow_file, mark.segment), []).append(mark) + stale: list[Mark] = [] + for line, line_marks in sorted(by_line.items()): + ordered = sorted(line_marks, key=lambda mark: mark.slot, reverse=True) + stale.extend(ordered if line not in current else ordered[KEEP_PER_LINE:]) + return stale def tick( repo: str, documents: Mapping[str, object], - state: dict[str, Handled], now: datetime, send: bool, + remote: Remote, + unreadable: Collection[str] = (), ) -> list[str]: - """Run one tick: update `state` in place and return the problems met. + """Run one tick against one repository and return the problems met. - A problem with one workflow or one dispatch is reported and the rest of the - tick goes on; the caller turns any problem into a failed run. + `unreadable` names the workflow files present on the default branch that + could not be read or parsed. A problem with one workflow or one dispatch is + reported and the rest of the tick goes on; the caller turns any problem into + a failed run. """ problems: list[str] = [] workflows: list[ScheduledWorkflow] = [] + unjudged = set(unreadable) for workflow_file, document in sorted(documents.items()): try: workflow = scheduled_workflow(workflow_file, document) except (MisconfiguredWorkflow, UnsupportedCron) as exc: problems.append(str(exc)) + unjudged.add(workflow_file) continue if workflow is not None: workflows.append(workflow) - for step in plan(workflows, state, now): + try: + marks = remote.read_marks(repo) + except (GitHubError, ValueError) as exc: + # With no marks there is no telling what was sent; sending anyway could + # repeat a slot, so this tick sends nothing. + problems.append(f"cannot read the clock's marks: {exc}") + return problems + for step in plan(workflows, marks, now): workflow_file = step.workflow.workflow_file - key = state_key(workflow_file, step.schedule) - handled = Handled(step.slot, step.workflow.enrolled) what = f"{workflow_file} cron {step.schedule.canonical!r} slot {format_slot(step.slot)}" - if step.kind == "baseline": - reason = "first seen" if key not in state else "enrollment changed" - print(f"clock: {what}: {reason}; baselined without sending") - state[key] = handled - continue + mark = mark_for(workflow_file, step.schedule, step.slot) if not send: - enrolled = "enrolled" if step.workflow.enrolled else "not enrolled" - print(f"clock: {what}: would send ({enrolled}; shadow mode)") - state[key] = handled - continue - if not step.workflow.enrolled: - state[key] = handled + verb = "would baseline" if step.kind == "baseline" else "would send" + print(f"clock: {what}: {verb} (dry run)") continue try: - if slot_ledger.is_done(repo, workflow_file, step.schedule, step.slot): + if not remote.claim(repo, mark): + print(f"clock: {what}: claimed by another clock; not sent") + continue + if step.kind == "baseline": + print( + f"clock: {what}: no clock has marked this line; baselined without sending" + ) + continue + if remote.is_done(repo, workflow_file, step.schedule, step.slot): print(f"clock: {what}: already recorded done; not sent") - else: - dispatch(repo, workflow_file, step.schedule, step.slot) - print(f"clock: {what}: sent") + continue except (GitHubError, ValueError) as exc: - # Left out of the state, so the next tick tries this slot again. A - # ValueError is the ledger refusing this workflow's file name or the - # repository; no retry fixes it, so it is reported every tick. problems.append(f"{what}: {exc}") continue - state[key] = handled + try: + remote.dispatch(repo, workflow_file, step.schedule, step.slot) + except (GitHubError, ValueError) as exc: + # The mark comes off again, so the next tick on any host sends the + # slot. A ValueError is a name no retry fixes; it is reported on + # every tick. + problems.append(f"{what}: {exc}") + try: + remote.release(repo, mark) + except (GitHubError, ValueError) as release_exc: + problems.append(f"{what}: releasing its mark failed: {release_exc}") + continue + print(f"clock: {what}: sent") + stale = stale_marks(workflows, marks, unjudged) + if stale and send: + try: + remote.remove(repo, stale) + except (GitHubError, ValueError) as exc: + # Housekeeping: a mark left behind only costs a listing entry, and + # the next tick tries again. + print(f"clock: removing {len(stale)} old mark(s) failed: {exc}") + elif stale: + print(f"clock: would remove {len(stale)} old mark(s) (dry run)") return problems @@ -328,16 +369,10 @@ def _parser() -> argparse.ArgumentParser: description="Deliver every enrolled workflow's due slots once.", ) parser.add_argument("--repo", required=True, help="owner/name") - parser.add_argument( - "--state", - required=True, - type=Path, - help="state file, one per repository (a cache; see the README)", - ) parser.add_argument( "--send", action="store_true", - help="dispatch due slots to enrolled workflows (default: shadow mode, log only)", + help="claim and dispatch due slots (default: a dry run, which writes nothing)", ) return parser @@ -345,7 +380,8 @@ def _parser() -> argparse.ArgumentParser: def main( argv: Sequence[str], now: datetime | None = None, - load: workflow_files.Loader = workflow_files.load_default_branch, + load: default_branch.Loader = default_branch.load, + remote: Remote | None = None, ) -> int: args = _parser().parse_args(argv) try: @@ -353,11 +389,11 @@ def main( except (GitHubError, ValueError, yq.ParserUnavailable) as exc: print(f"clock: cannot read any workflow of {args.repo}: {exc}") return 1 - state = load_state(args.state) problems = [f"{name}: unreadable: {reason}" for name, reason in unreadable.items()] current = now if now is not None else datetime.now(UTC) - problems += tick(args.repo, documents, state, current, args.send) - save_state(args.state, state) + problems += tick( + args.repo, documents, current, args.send, remote or GitHubRemote(), unreadable + ) for problem in problems: print(f"clock: problem: {problem}") return 1 if problems else 0 diff --git a/src/github_cron_trigger/clock_marks.py b/src/github_cron_trigger/clock_marks.py new file mode 100644 index 0000000..6649ee9 --- /dev/null +++ b/src/github_cron_trigger/clock_marks.py @@ -0,0 +1,112 @@ +"""The clock's record of what it has sent, kept in the served repository. + +One git ref per slot a clock claimed: + + refs/github-cron-trigger-clock/// + +in the ledger's own encoding (slot_ledger.py), pointing at a blob that says +which slot it marks. Nothing reads what a mark points at. + +WHY IN THE REPOSITORY. Kept on the host, a clock's memory ties delivery to that +one host: while it sleeps every slot waits for GitHub's backstop, and a second +host would send every slot again. Kept here, any number of hosts can tick the +same repository, a host can be replaced with nothing to copy, and the record +can be read from the repository it describes. + +WHY A CLAIM BEFORE THE SEND. Creating a ref is atomic: of the clocks that +create the same mark, exactly one succeeds and the rest are told it already +exists. A clock sends only after its create succeeded, so each slot is sent by +one clock however many tick at once. A send that fails releases its mark, so +the next tick on any host tries the slot again. A clock stopped between its +claim and its send leaves a slot marked and not sent; GitHub's schedule still +delivers that one, late. + +This is not the slot ledger. The ledger records slots whose work finished, +whichever delivery did it, and the slot action reads it; marks record slots a +clock sent, for every enrolled workflow, reconcilers included, and only clocks +read them. +""" + +import re +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import datetime + +from .cron_slots import CronSchedule, format_slot, ledger_key, parse_ledger_key +from .slot_ledger import create_ref, cron_segment, delete_ref, list_refs + +MARK_NAMESPACE = "github-cron-trigger-clock" + +# How many of a line's newest marks pruning keeps. Two, not one: a clock whose +# send fails releases its new mark, and the one before it must still be there, +# or the line would read as never sent and baseline instead of retrying. +KEEP_PER_LINE = 2 + +_WORKFLOW_FILE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\.ya?ml$") +_SEGMENT_FIELD = r"(?:x|\d{1,2}(?:,\d{1,2})*)" +_MARK = re.compile( + rf"^(?P[A-Za-z0-9][A-Za-z0-9._-]*\.ya?ml)" + rf"/(?P{_SEGMENT_FIELD}(?:_{_SEGMENT_FIELD}){{4}})" + r"/(?P\d{8}T\d{4}Z)$" +) + + +@dataclass(frozen=True, order=True) +class Mark: + """One slot a clock claimed: the workflow file, the cron line as its ref + segment (slot_ledger.cron_segment), and the slot.""" + + workflow_file: str + segment: str + slot: datetime + + @property + def path(self) -> str: + """The ref path under `refs/`.""" + return f"{MARK_NAMESPACE}/{self.workflow_file}/{self.segment}/{ledger_key(self.slot)}" + + +def mark_for(workflow_file: str, schedule: CronSchedule, slot: datetime) -> Mark: + if not _WORKFLOW_FILE.match(workflow_file): + raise ValueError(f"workflow file {workflow_file!r} is not a workflow basename") + return Mark(workflow_file, cron_segment(schedule), slot) + + +def parse_mark(name: str) -> Mark | None: + """The mark a ref name under the namespace holds, or None for a name the + clock did not write: another form, or a key naming no instant.""" + match = _MARK.match(name) + if match is None: + return None + try: + slot = parse_ledger_key(match.group("key")) + except ValueError: + return None + return Mark(match.group("file"), match.group("segment"), slot) + + +def note(mark: Mark) -> str: + """The text of a mark's blob. It names the slot and nothing about the host, + since a public repository's refs are readable by anyone.""" + return f"github-cron-trigger clock: claimed {mark.workflow_file} slot {format_slot(mark.slot)}\n" + + +class GitHubMarks: + """Marks read and written through the GitHub API.""" + + def read(self, repo: str) -> list[Mark]: + """Every mark the repository holds. A ref in the namespace that is not + in the clock's form is left out, and never deleted.""" + marks = [parse_mark(name) for name in list_refs(repo, f"{MARK_NAMESPACE}/")] + return sorted(mark for mark in marks if mark is not None) + + def claim(self, repo: str, mark: Mark) -> bool: + """Create the mark; True when this call created it.""" + return create_ref(repo, mark.path, note(mark)) + + def release(self, repo: str, mark: Mark) -> None: + delete_ref(repo, mark.path) + + def remove(self, repo: str, marks: Iterable[Mark]) -> None: + for mark in marks: + delete_ref(repo, mark.path) diff --git a/src/github_cron_trigger/default_branch.py b/src/github_cron_trigger/default_branch.py new file mode 100644 index 0000000..a467c56 --- /dev/null +++ b/src/github_cron_trigger/default_branch.py @@ -0,0 +1,323 @@ +"""The workflow files on a repository's default branch, read so that an idle +tick costs one small request. + +A clock ticks every few minutes and workflow files change a few times a day, +so nearly every tick would re-read what it read last time. A repository's +workflow files can run to megabytes, and each would need a yq run to parse. +Git names every tree and file version by a hash of its content, so a reader +can ask for that name first and reuse everything it already has under it: + +1. One query asks for the id of the default branch's `.github/workflows` tree. + That id changes only when a workflow file changes. +2. If the tree is new to this host, a second query lists its entries - names + and blob ids - by the tree's id, so they describe exactly that tree. +3. A file whose blob was parsed before reuses the parse. Otherwise its text + comes from the cache, or one batched query fetches every missing text, and + yq parses only those files. + +THE CACHE IS ONLY A CACHE. Everything in it is keyed by a content id, so it is +never stale: a changed file has a new id and misses. A cached text is checked +against its id by re-hashing it, and anything unreadable is a miss. Deleting +the cache costs one full read and changes no decision, so a host still keeps +nothing that matters (clock_marks.py). Entries unused for CACHE_MAX_AGE are +pruned whenever a read had to add something. + +A parse is keyed by blob id and PARSE_FORMAT, not by yq's version: the yq +floor (yq.MINIMUM_VERSION) is set where every supported release reads workflow +files alike, so an idle tick runs no yq at all. +""" + +import hashlib +import json +import os +import re +import tempfile +import time +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from . import workflow_files, yq +from .github import GitHubError, check_repo, gh_api + +# Bumped whenever the parse rules in workflow_files.parse change, or the yq +# floor moves, so every earlier parse misses. +PARSE_FORMAT = "1" + +# How long a cache entry nobody used is kept. A file version a repository no +# longer has is dead weight; one it still has is touched on every read. +CACHE_MAX_AGE = 30 * 24 * 3600 + +# Texts per batched query: well inside GitHub's GraphQL node limits. +_BLOB_BATCH = 50 + +_OID = re.compile(r"^[0-9a-f]{40}$") + +# Runs one GraphQL query: (query, variables) -> its `data`. Replaced in tests. +GraphQL = Callable[[str, Mapping[str, str]], Any] + +# Reads one repository: owner/name -> the parsed workflows by file name, and +# the files that could not be read or parsed, with the reason. +Loader = Callable[[str], tuple[dict[str, object], dict[str, str]]] + +_TREE_QUERY = """ +query($owner: String!, $name: String!) { + repository(owner: $owner, name: $name) { + object(expression: "HEAD:.github/workflows") { oid } + } +} +""" + +_ENTRIES_QUERY = """ +query($owner: String!, $name: String!, $tree: GitObjectID!) { + repository(owner: $owner, name: $name) { + object(oid: $tree) { ... on Tree { entries { name type oid } } } + } +} +""" + + +def graphql(query: str, variables: Mapping[str, str]) -> Any: + """Run one GraphQL query through `gh` and return its `data`.""" + arguments = ["graphql", "-f", f"query={query}"] + for name, value in variables.items(): + arguments += ["-f", f"{name}={value}"] + answer = json.loads(gh_api(arguments)) + if not isinstance(answer, dict) or not isinstance(answer.get("data"), dict): + raise GitHubError(f"GraphQL answered without data: {answer!r}"[:500]) + return answer["data"] + + +def git_blob_id(data: bytes) -> str: + """The id git gives a blob holding `data`.""" + header = b"blob %d\0" % len(data) + return hashlib.sha1(header + data, usedforsecurity=False).hexdigest() + + +@dataclass(frozen=True) +class Entry: + """One workflow file in a tree: its name and its blob's id.""" + + name: str + oid: str + + +class Cache: + """Tree listings, file texts and parses, on disk, keyed by content id. + Every failure here is a miss, never an error: nothing depends on it.""" + + def __init__(self, root: Path | None) -> None: + self.root = root + self.wrote = False + + def _path(self, kind: str, oid: str) -> Path | None: + if self.root is None or not _OID.match(oid): + return None + return self.root / kind / oid[:2] / oid + + def _read(self, path: Path | None) -> bytes | None: + if path is None: + return None + try: + data = path.read_bytes() + # Touched so pruning keeps what is still in use. + os.utime(path) + except OSError: + return None + return data + + def _write(self, path: Path | None, data: bytes) -> None: + if path is None: + return + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".tmp-") + with os.fdopen(fd, "wb") as handle: + handle.write(data) + os.replace(temporary, path) + except OSError: + return + self.wrote = True + + def entries(self, tree: str) -> list[Entry] | None: + data = self._read(self._path("trees", tree)) + try: + listed = json.loads(data) if data is not None else None + if not isinstance(listed, list): + return None + entries = [Entry(str(name), str(oid)) for name, oid in listed] + except (ValueError, TypeError): + return None + if not all(_OID.match(entry.oid) for entry in entries): + return None + return entries + + def put_entries(self, tree: str, entries: Sequence[Entry]) -> None: + listed = [[entry.name, entry.oid] for entry in entries] + self._write(self._path("trees", tree), json.dumps(listed).encode("utf-8")) + + def text(self, oid: str) -> str | None: + data = self._read(self._path("blobs", oid)) + if data is None or git_blob_id(data) != oid: + return None + try: + return data.decode("utf-8") + except UnicodeDecodeError: + return None + + def put_text(self, oid: str, text: str) -> None: + data = text.encode("utf-8") + # A text that does not hash to its id would be refused on every read. + if git_blob_id(data) == oid: + self._write(self._path("blobs", oid), data) + + def parsed(self, oid: str) -> tuple[object, str | None] | None: + """(document, None) or (None, the reason it did not parse).""" + data = self._read(self._path(f"parsed-{PARSE_FORMAT}", oid)) + try: + stored = json.loads(data) if data is not None else None + except ValueError: + return None + if not isinstance(stored, dict): + return None + if "error" in stored and isinstance(stored["error"], str): + return None, stored["error"] + if "document" in stored: + return stored["document"], None + return None + + def put_parsed(self, oid: str, document: object, error: str | None) -> None: + stored = {"error": error} if error is not None else {"document": document} + path = self._path(f"parsed-{PARSE_FORMAT}", oid) + self._write(path, json.dumps(stored).encode("utf-8")) + + def prune(self, now: float) -> None: + """Remove entries unused for CACHE_MAX_AGE.""" + if self.root is None: + return + try: + files = [path for path in self.root.rglob("*") if path.is_file()] + except OSError: + return + for path in files: + try: + if now - path.stat().st_mtime > CACHE_MAX_AGE: + path.unlink() + except OSError: + continue + + +def default_cache() -> Cache: + """The user cache directory's cache, shared by every repository a host + reads: content ids are the same in every repository.""" + xdg = os.environ.get("XDG_CACHE_HOME") + root = Path(xdg) if xdg else Path.home() / ".cache" + return Cache(root / "github-cron-trigger") + + +def _tree_entries(answer: Any) -> list[Entry]: + entries: list[Entry] = [] + for entry in answer["entries"]: + name, kind, oid = entry["name"], entry["type"], entry["oid"] + if kind != "blob" or not name.endswith(workflow_files.WORKFLOW_SUFFIXES): + continue + if not _OID.match(oid): + raise GitHubError(f"workflow file {name!r} answered blob id {oid!r}") + entries.append(Entry(name, oid)) + return sorted(entries, key=lambda entry: entry.name) + + +def _fetch_texts( + owner: str, name: str, oids: Sequence[str], call: GraphQL +) -> dict[str, str | None]: + """Each blob's text, or None where GitHub returned no complete text.""" + texts: dict[str, str | None] = {} + for start in range(0, len(oids), _BLOB_BATCH): + batch = oids[start : start + _BLOB_BATCH] + # The ids are interpolated, so each was matched against _OID first. + fields = " ".join( + f'b{index}: object(oid: "{oid}") {{ ... on Blob {{ text isBinary isTruncated }} }}' + for index, oid in enumerate(batch) + ) + query = ( + "query($owner: String!, $name: String!) {" + f" repository(owner: $owner, name: $name) {{ {fields} }} }}" + ) + node = call(query, {"owner": owner, "name": name})["repository"] + for index, oid in enumerate(batch): + blob = node.get(f"b{index}") + complete = ( + isinstance(blob, dict) + and isinstance(blob.get("text"), str) + and not blob.get("isBinary") + and not blob.get("isTruncated") + ) + texts[oid] = blob["text"] if complete else None + return texts + + +def load( + repo: str, + cache: Cache | None = None, + call: GraphQL = graphql, + executable: Callable[[], str] = yq.executable, + now: Callable[[], float] = time.time, +) -> tuple[dict[str, object], dict[str, str]]: + """Every workflow on the default branch, parsed, and the files that could + not be read or parsed, by name, with the reason.""" + store = cache if cache is not None else default_cache() + owner, name = check_repo(repo).split("/") + variables = {"owner": owner, "name": name} + try: + tree = call(_TREE_QUERY, variables)["repository"]["object"] + # A repository with no workflow directory has nothing to read. + if tree is None: + return {}, {} + tree_oid = tree["oid"] + if not isinstance(tree_oid, str) or not _OID.match(tree_oid): + raise GitHubError(f"the workflow tree answered id {tree_oid!r}") + entries = store.entries(tree_oid) + if entries is None: + listed = call(_ENTRIES_QUERY, {**variables, "tree": tree_oid}) + entries = _tree_entries(listed["repository"]["object"]) + store.put_entries(tree_oid, entries) + documents: dict[str, object] = {} + unreadable: dict[str, str] = {} + unparsed: list[Entry] = [] + for entry in entries: + parsed = store.parsed(entry.oid) + if parsed is None: + unparsed.append(entry) + elif parsed[1] is not None: + unreadable[entry.name] = parsed[1] + else: + documents[entry.name] = parsed[0] + if unparsed: + texts = {entry.oid: store.text(entry.oid) for entry in unparsed} + missing = [oid for oid, text in texts.items() if text is None] + if missing: + texts.update(_fetch_texts(owner, name, missing, call)) + parser = executable() + for entry in unparsed: + text = texts.get(entry.oid) + if text is None: + # Not cached: GitHub may return the whole text next time. + unreadable[entry.name] = "GitHub returned no complete text for it" + continue + store.put_text(entry.oid, text) + try: + document: object = workflow_files.parse(text, parser) + except workflow_files.UnreadableWorkflow as exc: + store.put_parsed(entry.oid, None, str(exc)) + unreadable[entry.name] = str(exc) + continue + store.put_parsed(entry.oid, document, None) + documents[entry.name] = document + except (KeyError, TypeError) as exc: + raise GitHubError( + f"reading {repo}'s workflow files: an answer in an unexpected shape: {exc!r}" + ) from exc + if store.wrote: + store.prune(now()) + return documents, unreadable diff --git a/src/github_cron_trigger/freshness.py b/src/github_cron_trigger/freshness.py index 89a72a6..4dbc175 100644 --- a/src/github_cron_trigger/freshness.py +++ b/src/github_cron_trigger/freshness.py @@ -48,7 +48,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta -from . import slot_ledger, workflow_files, yq +from . import default_branch, slot_ledger, yq from .clock import EVENT_TYPE_PREFIX, scheduled_workflow, uses_slot_action from .cron_slots import ( UTC, @@ -341,7 +341,7 @@ def main( argv: Sequence[str], now: datetime | None = None, records: Callable[[str, str], list[str]] = slot_ledger.records, - load: workflow_files.Loader = workflow_files.load_default_branch, + load: default_branch.Loader = default_branch.load, ) -> int: parser = argparse.ArgumentParser( prog="github-cron-trigger missed", diff --git a/src/github_cron_trigger/slot_ledger.py b/src/github_cron_trigger/slot_ledger.py index 35e1c49..2e8aef5 100644 --- a/src/github_cron_trigger/slot_ledger.py +++ b/src/github_cron_trigger/slot_ledger.py @@ -128,15 +128,25 @@ def record( True when this call created the record, False when it already existed - another delivery of the same slot got there first, or this call's own retry - found its earlier attempt applied. Either way the slot is recorded. - - The blob is written first, so a failed ref create can leave it behind, - unreferenced and inert. Writing the same note again names the same blob, - which is what makes a retried call harmless. + found its earlier attempt applied (create_ref). Either way the slot is + recorded. """ if not _OBJECT.match(sha): raise ValueError(f"commit {sha!r} is not a full object name") - path = ref_path(workflow_file, schedule, slot, part) + return create_ref( + repo, ref_path(workflow_file, schedule, slot, part), record_note(sha) + ) + + +def create_ref(repo: str, path: str, note: str) -> bool: + """Create `refs/` pointing at a blob holding `note`, atomically. + + True when this call created the ref, False when it already existed - another + caller got there first, or this call's own retry found its earlier attempt + applied. The blob is written first, so a failed ref create can leave it + behind, unreferenced and inert. Writing the same note again names the same + blob, which is what makes a retried call harmless. + """ repo = check_repo(repo) blob = gh_api( [ @@ -144,7 +154,7 @@ def record( "POST", f"repos/{repo}/git/blobs", "-f", - f"content={record_note(sha)}", + f"content={note}", "-f", "encoding=utf-8", "--jq", @@ -172,39 +182,52 @@ def record( return True -def records(repo: str, workflow_file: str) -> list[str]: - """Every record one workflow holds, as `/`, sorted.""" - prefix = f"refs/{REF_NAMESPACE}/{_check_workflow_file(workflow_file)}/" - # The trailing '/' keeps a prefix match from reaching a workflow whose name - # merely starts with this one. +def delete_ref(repo: str, path: str) -> None: + """Delete `refs/`. An already-absent ref is not an error.""" + try: + gh_api(["--method", "DELETE", f"repos/{check_repo(repo)}/git/refs/{path}"]) + except HttpError as exc: + # GitHub answers a delete of a missing ref with 422 "Reference does not + # exist" rather than a 404; any other 422 is a real failure. + missing = exc.status == 404 or ( + exc.status == 422 and "does not exist" in exc.message.lower() + ) + if not missing: + raise + + +def list_refs(repo: str, prefix: str) -> list[str]: + """Every ref under `refs/`, as the part after the prefix, sorted. + `prefix` ends in '/', so a prefix match cannot reach a sibling whose name + merely starts with the last segment.""" + if not prefix.endswith("/"): + raise ValueError(f"ref prefix {prefix!r} must end in '/'") out = gh_api( [ "--paginate", - f"repos/{check_repo(repo)}/git/matching-refs/{prefix[len('refs/') :]}", + f"repos/{check_repo(repo)}/git/matching-refs/{prefix}", "--jq", ".[].ref", ] ) + full = f"refs/{prefix}" return sorted( - line[len(prefix) :] for line in out.splitlines() if line.startswith(prefix) + line[len(full) :] for line in out.splitlines() if line.startswith(full) ) +def records(repo: str, workflow_file: str) -> list[str]: + """Every record one workflow holds, as `/`, sorted.""" + return list_refs(repo, f"{REF_NAMESPACE}/{_check_workflow_file(workflow_file)}/") + + def delete(repo: str, workflow_file: str, record_name: str) -> None: """Remove one record. An already-absent record is not an error.""" if not _RECORD.match(record_name): raise ValueError(f"record {record_name!r} is not a ledger record") - path = f"{REF_NAMESPACE}/{_check_workflow_file(workflow_file)}/{record_name}" - try: - gh_api(["--method", "DELETE", f"repos/{check_repo(repo)}/git/refs/{path}"]) - except HttpError as exc: - # GitHub answers a delete of a missing ref with 422 "Reference does not - # exist" rather than a 404; any other 422 is a real failure. - missing = exc.status == 404 or ( - exc.status == 422 and "does not exist" in exc.message.lower() - ) - if not missing: - raise + delete_ref( + repo, f"{REF_NAMESPACE}/{_check_workflow_file(workflow_file)}/{record_name}" + ) def record_slot_key(record_name: str) -> str | None: diff --git a/src/github_cron_trigger/workflow_files.py b/src/github_cron_trigger/workflow_files.py index 30596a9..6178f92 100644 --- a/src/github_cron_trigger/workflow_files.py +++ b/src/github_cron_trigger/workflow_files.py @@ -1,8 +1,9 @@ """Workflow files, read from GitHub and parsed into the document GitHub reads. -Every reader here works at a commit the caller names or at the default branch, -never from a checkout: the slot action needs no checkout of the repository it -runs in, and the clock can serve a repository it has no clone of. +The slot action reads one file at the commit its run executed (fetch_file), +never from a checkout, so it needs no checkout of the repository it runs in. +The clock's and the missed-slot check's reads of the default branch are +default_branch.py's. PARSING goes through mikefarah's yq on PATH (yq.py), whose JSON output Python reads. A file is refused rather than half-read when it is not one workflow GitHub would @@ -14,11 +15,11 @@ import json import re import subprocess -from collections.abc import Callable, Mapping +from collections.abc import Mapping from typing import Any from . import yq -from .github import GitHubError, check_repo, gh_api +from .github import check_repo, gh_api WORKFLOW_SUFFIXES = (".yml", ".yaml") WORKFLOW_DIRECTORY = ".github/workflows" @@ -45,29 +46,6 @@ # A bound on one parse, so a wedged yq cannot hold a run open. _YQ_TIMEOUT_SECONDS = 60 -# Every workflow file of the default branch in one query, however many there -# are; reading them one REST call at a time would cost a call per file per tick. -_DEFAULT_BRANCH_QUERY = """ -query($owner: String!, $name: String!) { - repository(owner: $owner, name: $name) { - object(expression: "HEAD:.github/workflows") { - ... on Tree { - entries { - name - type - object { ... on Blob { text isBinary isTruncated } } - } - } - } - } -} -""" - - -# Reads every workflow of a repository's default branch: owner/name -> the parsed -# documents by file name, and the files that could not be read, with the reason. -Loader = Callable[[str], tuple[dict[str, object], dict[str, str]]] - class UnreadableWorkflow(ValueError): """A workflow file that cannot be read as one workflow document.""" @@ -151,56 +129,3 @@ def fetch_file(repo: str, path: str, commit: str) -> str: f"repos/{check_repo(repo)}/contents/{check_workflow_path(path)}?ref={commit}", ] ) - - -def fetch_default_branch(repo: str) -> tuple[dict[str, str], dict[str, str]]: - """The text of every workflow file on the default branch, keyed by file - name, and the files whose text GitHub would not return, with the reason.""" - owner, name = check_repo(repo).split("/") - answer = json.loads( - gh_api( - [ - "graphql", - "-f", - f"query={_DEFAULT_BRANCH_QUERY}", - "-f", - f"owner={owner}", - "-f", - f"name={name}", - ] - ) - ) - texts: dict[str, str] = {} - unreadable: dict[str, str] = {} - try: - tree = answer["data"]["repository"]["object"] - # A repository with no workflow directory has nothing to read. - entries = [] if tree is None else tree["entries"] - for entry in entries: - file_name = entry["name"] - if entry["type"] != "blob" or not file_name.endswith(WORKFLOW_SUFFIXES): - continue - blob = entry["object"] - if blob["isBinary"] or blob["isTruncated"] or blob["text"] is None: - unreadable[file_name] = "GitHub returned no complete text for it" - else: - texts[file_name] = blob["text"] - except (KeyError, TypeError) as exc: - raise GitHubError( - f"the workflow files query answered in an unexpected shape: {exc!r}" - ) from exc - return texts, unreadable - - -def load_default_branch( - repo: str, - fetch: Callable[ - [str], tuple[dict[str, str], dict[str, str]] - ] = fetch_default_branch, - executable: Callable[[], str] = yq.executable, -) -> tuple[dict[str, object], dict[str, str]]: - """Every workflow on the default branch, parsed, and the files that could - not be read or parsed, with the reason.""" - texts, unfetched = fetch(repo) - documents, unparsed = parse_all(texts, executable()) - return documents, {**unfetched, **unparsed} diff --git a/tests/live_check.py b/tests/live_check.py index ae1808f..01b1b2b 100644 --- a/tests/live_check.py +++ b/tests/live_check.py @@ -24,21 +24,39 @@ THE READERS. The slot command reads the running workflow's own file through the contents API at the run's commit and parses it with the runner's yq; the clock and the missed-slot check read every workflow of the default branch in one -GraphQL query. Each is run here as a step or a host would run it: a dispatched slot of this workflow's own line resolves and reads -the ledger, a line the workflow does not carry is refused, a shadow-mode tick -baselines this repository's lines without sending, and the missed-slot check -reads everything. +GraphQL query. Each is run here as a step or a host would run it: a dispatched +slot of this workflow's own line resolves and reads the ledger, a line the +workflow does not carry is refused, and the missed-slot check reads everything. + +THE CLOCKS. Two clock processes tick this repository with --send at the same +moment, as two hosts would, against clock-check.yml, a workflow enrolled for +this. With no marks, they baseline its newest slot once and send nothing; with +an older slot marked, exactly one of them sends the newest. The marks live in +the repository (clock_marks.py), and are removed again whatever the outcome. """ import argparse import contextlib import io +import os +import subprocess import sys import tempfile +from collections.abc import Mapping from datetime import datetime, timedelta from pathlib import Path +from typing import Any -from github_cron_trigger import clock, freshness, slot_ledger, steps, workflow_files +from github_cron_trigger import ( + clock, + default_branch, + freshness, + slot_ledger, + steps, + workflow_files, + yq, +) +from github_cron_trigger.clock_marks import GitHubMarks, Mark, mark_for from github_cron_trigger.cron_slots import UTC, format_slot, latest_slot, parse_cron from github_cron_trigger.github import gh_api from github_cron_trigger.workflow_triggers import utc_schedule @@ -183,12 +201,25 @@ def readers(repo: str, workflow_ref: str, workflow_sha: str) -> None: "a dispatch of a line this workflow does not carry is refused", ) with tempfile.TemporaryDirectory() as tmp: - state = Path(tmp) / "state.json" - status = clock.main(["--repo", repo, "--state", str(state)]) - handled = clock.load_state(state) + cache = default_branch.Cache(Path(tmp)) + calls: list[str] = [] + parses: list[str] = [] + + def counted(query: str, variables: Mapping[str, str]) -> Any: + calls.append(query) + return default_branch.graphql(query, variables) + + def parser() -> str: + parses.append("yq") + return yq.executable() + + first = default_branch.load(repo, cache, counted, parser) + calls.clear() + parses.clear() + second = default_branch.load(repo, cache, counted, parser) _expect( - status == 0 and bool(handled), - f"a shadow-mode tick reads the default branch and baselines {len(handled)} line(s)", + second == first and len(calls) == 1 and not parses, + f"an idle read of {len(first[0])} workflow file(s) is one query and no parse", ) status = freshness.main(["--repo", repo]) _expect( @@ -197,6 +228,102 @@ def readers(repo: str, workflow_ref: str, workflow_sha: str) -> None: ) +# The enrolled workflow the clock check runs real clocks against. Its yearly line +# always has a newest slot, and the slot before it, to claim; the workflow says +# why it exists. +CLOCK_CHECK = "clock-check.yml" +YEARLY = parse_cron("0 0 1 1 *") +MARKS = GitHubMarks() + + +def _clock_marks(repo: str) -> list[Mark]: + return [mark for mark in MARKS.read(repo) if mark.workflow_file == CLOCK_CHECK] + + +def _concurrent_ticks(repo: str) -> list[str]: + """Two clocks ticking the repository with --send at the same moment, as two + hosts would; their output.""" + environment = { + **os.environ, + "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src"), + } + command = [ + sys.executable, + "-m", + "github_cron_trigger", + "tick", + "--repo", + repo, + "--send", + ] + clocks = [ + subprocess.Popen(command, stdout=subprocess.PIPE, text=True, env=environment) + for _ in range(2) + ] + outputs = [clock_process.communicate(timeout=600)[0] for clock_process in clocks] + for output in outputs: + print(output, end="") + return outputs + + +def _default_branch_has(repo: str, workflow_file: str) -> bool: + names = gh_api([f"repos/{repo}/contents/.github/workflows", "--jq", ".[].name"]) + return workflow_file in names.split() + + +def _sends(outputs: list[str]) -> int: + return sum( + line.startswith(f"clock: {CLOCK_CHECK} cron") and line.endswith(": sent") + for output in outputs + for line in output.splitlines() + ) + + +def clocks(repo: str, required: bool) -> None: + """Two clocks at once: a line no clock has marked is baselined once and + sent nothing, and a due slot is sent exactly once. + + The clock reads the default branch, so the pull request that adds or changes + clock-check.yml cannot run this against its own copy. A pull request whose + default branch lacks the workflow skips it, saying so; any other run + (`required`) fails instead, so the default branch always runs it. + """ + if not _default_branch_has(repo, CLOCK_CHECK): + if required: + raise AssertionError(f"{CLOCK_CHECK} is not on the default branch") + print(f"skipped: {CLOCK_CHECK} is not on the default branch yet") + return + MARKS.remove(repo, _clock_marks(repo)) + now = datetime.now(UTC) + newest = latest_slot(YEARLY, now) + printed = io.StringIO() + with contextlib.redirect_stdout(printed): + status = clock.main(["--repo", repo]) + _expect( + status == 0 + and "would baseline (dry run)" in printed.getvalue() + and not _clock_marks(repo), + "a dry run reads the repository, reports a baseline, and writes nothing", + ) + outputs = _concurrent_ticks(repo) + _expect( + _sends(outputs) == 0 and [mark.slot for mark in _clock_marks(repo)] == [newest], + "two clocks baselining at once mark the newest slot once and send nothing", + ) + MARKS.remove(repo, _clock_marks(repo)) + MARKS.claim( + repo, + mark_for( + CLOCK_CHECK, YEARLY, latest_slot(YEARLY, newest - timedelta(minutes=1)) + ), + ) + outputs = _concurrent_ticks(repo) + _expect( + _sends(outputs) == 1 and newest in [mark.slot for mark in _clock_marks(repo)], + "two clocks ticking a due slot at once send it exactly once", + ) + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", required=True, help="owner/name") @@ -206,8 +333,17 @@ def main() -> int: ) parser.add_argument("--workflow-ref", required=True, help="GITHUB_WORKFLOW_REF") parser.add_argument("--workflow-sha", required=True, help="GITHUB_WORKFLOW_SHA") + parser.add_argument( + "--require-clock-check", + action="store_true", + help=f"fail rather than skip when {CLOCK_CHECK} is not on the default branch", + ) args = parser.parse_args() readers(args.repo, args.workflow_ref, args.workflow_sha) + try: + clocks(args.repo, args.require_clock_check) + finally: + MARKS.remove(args.repo, _clock_marks(args.repo)) try: run(args.repo, args.sha, args.workflow_file) finally: diff --git a/tests/test_clock.py b/tests/test_clock.py index c52a427..4e61c9d 100644 --- a/tests/test_clock.py +++ b/tests/test_clock.py @@ -5,33 +5,39 @@ nothing listens for) or missed when it is (it would never be delivered on time); a first tick, or the first tick after an enrollment, that dispatches a slot the workflow ran before it could record it; a wake after a long sleep that -bursts every missed slot; and a lost state file that sends instead of -re-baselining. They also pin the refusal of a slot-action workflow enrolled on a -line firing more than once a day, whose every backstop run would fail. Workflow -files are real YAML read through the same reader the clock uses, not hand-built -mappings of what a parser returns. Every instant is hardcoded. +bursts every missed slot; two clocks ticking at once that both send a slot; a +failed send that leaves its slot marked, so no clock retries it; and clean-up +that removes the marks a line needs. They also pin the refusal of a slot-action +workflow enrolled on a line firing more than once a day, whose every backstop +run would fail. Workflow files are real YAML read through the same reader the +clock uses, not hand-built mappings of what a parser returns. The repository's +marks, ledger and dispatches are an in-memory stand-in whose claims are atomic, +as GitHub's ref creates are. Every instant is hardcoded. """ import contextlib import io -import tempfile import unittest +from collections.abc import Iterable from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo -from github_cron_trigger import clock, workflow_files, yq +from github_cron_trigger import clock, default_branch, workflow_files, yq from github_cron_trigger.clock import ( - Handled, ScheduledWorkflow, - load_state, plan, - save_state, scheduled_workflow, - state_key, + stale_marks, tick, ) -from github_cron_trigger.cron_slots import UTC, UnsupportedCron, parse_cron +from github_cron_trigger.clock_marks import Mark, mark_for +from github_cron_trigger.cron_slots import ( + UTC, + CronSchedule, + UnsupportedCron, + parse_cron, +) from github_cron_trigger.github import GitHubError from github_cron_trigger.workflow_triggers import MisconfiguredWorkflow @@ -215,64 +221,60 @@ def test_every_workflow_in_this_repository_is_readable(self) -> None: DAILY = ScheduledWorkflow("backup.yml", (parse_cron("30 19 * * *"),), True) -DAILY_KEY = state_key("backup.yml", DAILY.schedules[0]) +LINE = DAILY.schedules[0] WEEKLY_AND_DAILY = ScheduledWorkflow( "two-lines.yml", (parse_cron("0 1 * * *"), parse_cron("0 1 * * 1")), True ) +REPO = "owner/name" + + +def marked(workflow: ScheduledWorkflow, *slots: datetime, line: int = 0) -> list[Mark]: + return [ + mark_for(workflow.workflow_file, workflow.schedules[line], s) for s in slots + ] class PlanTest(unittest.TestCase): - def test_a_line_seen_for_the_first_time_is_baselined_not_delivered(self) -> None: - # Its latest slot came due before the clock was watching; the workflow + def test_a_line_no_clock_has_marked_is_baselined_not_delivered(self) -> None: + # Its latest slot came due before any clock served it; the workflow # may already have run it, before any ledger record of it existed. - steps = plan([DAILY], {}, utc(2026, 9, 28, 20, 0)) + steps = plan([DAILY], [], utc(2026, 9, 28, 20, 0)) self.assertEqual( [(s.kind, s.slot) for s in steps], [("baseline", utc(2026, 9, 28, 19, 30))] ) - def test_a_handled_slot_is_not_owed_again(self) -> None: - state = {DAILY_KEY: Handled(utc(2026, 9, 28, 19, 30), True)} - self.assertEqual(plan([DAILY], state, utc(2026, 9, 28, 23, 0)), []) + def test_a_marked_slot_is_not_owed_again(self) -> None: + marks = marked(DAILY, utc(2026, 9, 28, 19, 30)) + self.assertEqual(plan([DAILY], marks, utc(2026, 9, 28, 23, 0)), []) def test_a_new_slot_is_delivered_when_it_comes_due(self) -> None: - state = {DAILY_KEY: Handled(utc(2026, 9, 27, 19, 30), True)} - self.assertEqual(plan([DAILY], state, utc(2026, 9, 28, 19, 29)), []) - steps = plan([DAILY], state, utc(2026, 9, 28, 19, 30)) + marks = marked(DAILY, utc(2026, 9, 27, 19, 30)) + self.assertEqual(plan([DAILY], marks, utc(2026, 9, 28, 19, 29)), []) + steps = plan([DAILY], marks, utc(2026, 9, 28, 19, 30)) self.assertEqual( [(s.kind, s.slot) for s in steps], [("deliver", utc(2026, 9, 28, 19, 30))] ) def test_a_long_sleep_owes_the_newest_slot_only(self) -> None: - # Three days asleep: one delivery of the newest slot, not a burst of - # every missed one; the older ones are left to GitHub's backstop. - state = {DAILY_KEY: Handled(utc(2026, 9, 25, 19, 30), True)} - steps = plan([DAILY], state, utc(2026, 9, 28, 20, 0)) + # Three days with no clock up: one delivery of the newest slot, not a + # burst of every missed one; the older ones are left to GitHub's backstop. + marks = marked(DAILY, utc(2026, 9, 25, 19, 30)) + steps = plan([DAILY], marks, utc(2026, 9, 28, 20, 0)) self.assertEqual( [(s.kind, s.slot) for s in steps], [("deliver", utc(2026, 9, 28, 19, 30))] ) - def test_a_line_whose_workflow_enrolled_since_it_was_handled_is_baselined( - self, - ) -> None: - # The clock handled the 27th's slot while the workflow was not enrolled, - # then slept across the merge that enrolled it. The 28th's slot may have - # run before that merge, without the slot action and so with no ledger - # record, and a dispatch would repeat it. - state = {DAILY_KEY: Handled(utc(2026, 9, 27, 19, 30), False)} - steps = plan([DAILY], state, utc(2026, 9, 28, 21, 0)) - self.assertEqual( - [(s.kind, s.slot) for s in steps], [("baseline", utc(2026, 9, 28, 19, 30))] - ) + def test_a_workflow_that_is_not_enrolled_is_owed_nothing(self) -> None: + tracked = ScheduledWorkflow("backup.yml", DAILY.schedules, False) + self.assertEqual(plan([tracked], [], utc(2026, 9, 28, 20, 0)), []) def test_each_line_of_a_workflow_is_tracked_on_its_own(self) -> None: # Monday 2026-09-28 01:00 is a slot of each line; each is owed its own - # delivery, and the daily line's state says nothing about the weekly's. - daily, monday = WEEKLY_AND_DAILY.schedules - state = { - state_key("two-lines.yml", daily): Handled(utc(2026, 9, 27, 1, 0), True), - state_key("two-lines.yml", monday): Handled(utc(2026, 9, 21, 1, 0), True), - } - steps = plan([WEEKLY_AND_DAILY], state, utc(2026, 9, 28, 1, 2)) + # delivery, and the daily line's mark says nothing about the weekly's. + marks = marked(WEEKLY_AND_DAILY, utc(2026, 9, 27, 1, 0)) + marked( + WEEKLY_AND_DAILY, utc(2026, 9, 21, 1, 0), line=1 + ) + steps = plan([WEEKLY_AND_DAILY], marks, utc(2026, 9, 28, 1, 2)) self.assertEqual( [(s.schedule.canonical, s.slot) for s in steps], [ @@ -285,118 +287,240 @@ def test_utc_date_ahead_of_local_date(self) -> None: # 18:05 Pacific on the 27th is 01:05 UTC on the 28th: the 01:00 UTC # line's 28th slot is due, whatever the host's own calendar says. line = ScheduledWorkflow("logs.yml", (parse_cron("0 1 * * *"),), True) - state = { - state_key("logs.yml", line.schedules[0]): Handled( - utc(2026, 9, 27, 1, 0), True - ) - } now = datetime(2026, 9, 27, 18, 5, tzinfo=ZoneInfo("America/Los_Angeles")) - steps = plan([line], state, now) + steps = plan([line], marked(line, utc(2026, 9, 27, 1, 0)), now) self.assertEqual([s.slot for s in steps], [utc(2026, 9, 28, 1, 0)]) -class TickTest(unittest.TestCase): - def test_an_enrollment_is_delivered_from_the_slot_after_its_baseline( +class StaleMarksTest(unittest.TestCase): + def test_a_line_keeps_its_two_newest_marks(self) -> None: + # Two, so a clock whose send fails can release its new mark and still + # leave one behind to retry from. + marks = marked(DAILY, *(utc(2026, 9, day, 19, 30) for day in (25, 26, 27, 28))) + self.assertEqual( + [m.slot for m in stale_marks([DAILY], marks)], + [utc(2026, 9, 26, 19, 30), utc(2026, 9, 25, 19, 30)], + ) + + def test_a_line_no_enrolled_workflow_carries_loses_every_mark(self) -> None: + # Disenrolled, its cron changed, or its file deleted: enrolling it + # again later baselines it rather than sending. + marks = marked(DAILY, utc(2026, 9, 28, 19, 30)) + disenrolled = ScheduledWorkflow("backup.yml", DAILY.schedules, False) + self.assertEqual(stale_marks([disenrolled], marks), marks) + self.assertEqual(stale_marks([], marks), marks) + + def test_a_file_that_could_not_be_judged_keeps_its_marks(self) -> None: + # A file that failed to parse this tick is not a disenrolled one; its + # lines must deliver, not baseline, once it parses again. + marks = marked(DAILY, utc(2026, 9, 27, 19, 30), utc(2026, 9, 28, 19, 30)) + self.assertEqual(stale_marks([], marks, unjudged={"backup.yml"}), []) + + +class FakeRemote: + """A repository's marks, ledger and dispatches, in memory. A claim of a mark + that exists fails, as GitHub's ref create does.""" + + def __init__( self, + marks: Iterable[Mark] = (), + done: Iterable[datetime] = (), + failing_dispatches: int = 0, ) -> None: - # The baseline for a changed enrollment must record the line as handled - # while enrolled; otherwise every later slot would baseline again and - # the enrolled workflow would never be delivered. - enrolled = parsed({"backup.yml": ENROLLED}) - state: dict[str, Handled] = {} - with contextlib.redirect_stdout(io.StringIO()): - problems = tick( - "owner/name", - parsed({"backup.yml": TRACKED}), - state, - utc(2026, 9, 27, 20, 0), - send=False, - ) - problems += tick( - "owner/name", enrolled, state, utc(2026, 9, 28, 21, 0), send=False - ) + self.marks = set(marks) + self.done = set(done) + self.failing_dispatches = failing_dispatches + self.dispatched: list[tuple[str, datetime]] = [] + self.snapshot: list[Mark] | None = None + + def read_marks(self, repo: str) -> list[Mark]: + if self.snapshot is not None: + return list(self.snapshot) + return sorted(self.marks) + + def claim(self, repo: str, mark: Mark) -> bool: + if mark in self.marks: + return False + self.marks.add(mark) + return True + + def release(self, repo: str, mark: Mark) -> None: + self.marks.discard(mark) + + def remove(self, repo: str, marks: Iterable[Mark]) -> None: + self.marks.difference_update(marks) + + def is_done( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> bool: + return slot in self.done + + def dispatch( + self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime + ) -> None: + if self.failing_dispatches: + self.failing_dispatches -= 1 + raise GitHubError("gh: Server Error (HTTP 502)") + self.dispatched.append((workflow_file, slot)) + + +def run_tick( + remote: FakeRemote, + documents: dict[str, object], + now: datetime, + send: bool = True, + unreadable: Iterable[str] = (), +) -> tuple[list[str], str]: + printed = io.StringIO() + with contextlib.redirect_stdout(printed): + problems = tick(REPO, documents, now, send, remote, list(unreadable)) + return problems, printed.getvalue() + + +class TickTest(unittest.TestCase): + def setUp(self) -> None: + self.enrolled = parsed({"backup.yml": ENROLLED}) + + def test_a_due_slot_is_claimed_then_sent(self) -> None: + remote = FakeRemote(marked(DAILY, utc(2026, 9, 27, 19, 30))) + problems, printed = run_tick(remote, self.enrolled, utc(2026, 9, 28, 19, 31)) self.assertEqual(problems, []) - self.assertEqual(state, {DAILY_KEY: Handled(utc(2026, 9, 28, 19, 30), True)}) - workflow = scheduled_workflow("backup.yml", enrolled["backup.yml"]) - assert workflow is not None - steps = plan([workflow], state, utc(2026, 9, 29, 19, 30)) + self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 28, 19, 30))]) + self.assertIn( + mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30)), remote.marks + ) + self.assertIn(": sent", printed) + + def test_two_clocks_ticking_at_once_send_a_slot_once(self) -> None: + # Both read the marks before either claims: the race the claim exists + # for. Exactly one claim succeeds, and only that clock sends. + remote = FakeRemote(marked(DAILY, utc(2026, 9, 27, 19, 30))) + remote.snapshot = remote.read_marks(REPO) + now = utc(2026, 9, 28, 19, 31) + first, _ = run_tick(remote, self.enrolled, now) + second, printed = run_tick(remote, self.enrolled, now) + self.assertEqual(first + second, []) + self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 28, 19, 30))]) + self.assertIn("claimed by another clock; not sent", printed) + + def test_two_clocks_baselining_at_once_send_nothing(self) -> None: + remote = FakeRemote() + remote.snapshot = [] + now = utc(2026, 9, 28, 19, 31) + run_tick(remote, self.enrolled, now) + run_tick(remote, self.enrolled, now) + self.assertEqual(remote.dispatched, []) self.assertEqual( - [(s.kind, s.slot) for s in steps], [("deliver", utc(2026, 9, 29, 19, 30))] + remote.marks, {mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30))} + ) + + def test_a_failed_send_releases_its_mark_so_the_next_tick_retries(self) -> None: + remote = FakeRemote( + marked(DAILY, utc(2026, 9, 27, 19, 30)), failing_dispatches=1 + ) + now = utc(2026, 9, 28, 19, 31) + problems, _ = run_tick(remote, self.enrolled, now) + self.assertEqual(len(problems), 1) + self.assertNotIn( + mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30)), remote.marks + ) + # The previous mark is still there, so the line delivers rather than + # baselining. + problems, _ = run_tick(remote, self.enrolled, utc(2026, 9, 28, 19, 36)) + self.assertEqual(problems, []) + self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 28, 19, 30))]) + + def test_a_slot_the_ledger_records_is_marked_and_not_sent(self) -> None: + remote = FakeRemote( + marked(DAILY, utc(2026, 9, 27, 19, 30)), done=[utc(2026, 9, 28, 19, 30)] + ) + _, printed = run_tick(remote, self.enrolled, utc(2026, 9, 28, 19, 31)) + self.assertEqual(remote.dispatched, []) + self.assertIn( + mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30)), remote.marks + ) + self.assertIn("already recorded done; not sent", printed) + + def test_an_enrollment_is_delivered_from_the_slot_after_its_baseline( + self, + ) -> None: + # Not enrolled, the workflow has no marks; enrolled, its first tick + # baselines and the next slot is sent. + remote = FakeRemote() + run_tick(remote, parsed({"backup.yml": TRACKED}), utc(2026, 9, 27, 20, 0)) + self.assertEqual(remote.marks, set()) + run_tick(remote, self.enrolled, utc(2026, 9, 28, 21, 0)) + run_tick(remote, self.enrolled, utc(2026, 9, 29, 19, 31)) + self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 29, 19, 30))]) + + def test_disenrolling_removes_the_marks(self) -> None: + remote = FakeRemote(marked(DAILY, utc(2026, 9, 28, 19, 30))) + run_tick(remote, parsed({"backup.yml": TRACKED}), utc(2026, 9, 28, 20, 0)) + self.assertEqual(remote.marks, set()) + + def test_an_unreadable_file_keeps_its_marks(self) -> None: + marks = marked(DAILY, utc(2026, 9, 28, 19, 30)) + remote = FakeRemote(marks) + run_tick(remote, {}, utc(2026, 9, 28, 20, 0), unreadable=["backup.yml"]) + self.assertEqual(remote.marks, set(marks)) + + def test_a_dry_run_writes_nothing(self) -> None: + marks = marked(DAILY, *(utc(2026, 9, day, 19, 30) for day in (25, 26, 27))) + remote = FakeRemote(marks) + _, printed = run_tick( + remote, self.enrolled, utc(2026, 9, 28, 19, 31), send=False ) + self.assertEqual((remote.marks, remote.dispatched), (set(marks), [])) + self.assertIn("would send (dry run)", printed) + self.assertIn("would remove 1 old mark(s) (dry run)", printed) + + def test_marks_that_cannot_be_read_send_nothing(self) -> None: + class Unreadable(FakeRemote): + def read_marks(self, repo: str) -> list[Mark]: + raise GitHubError("gh: Server Error (HTTP 502)") + + remote = Unreadable() + problems, _ = run_tick(remote, self.enrolled, utc(2026, 9, 28, 19, 31)) + self.assertEqual(len(problems), 1) + self.assertEqual((remote.marks, remote.dispatched), (set(), [])) class MainTest(unittest.TestCase): """A tick that could not read what it had to deliver must fail, so whatever - runs the clock reports it, and must not lose the state it already holds.""" + runs the clock reports it; a file it could not read must not stop the rest.""" def run_main( - self, load: workflow_files.Loader, state: dict[str, Handled] - ) -> tuple[int, str, dict[str, Handled]]: - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "state.json" - save_state(path, state) - printed = io.StringIO() - with contextlib.redirect_stdout(printed): - status = clock.main( - ["--repo", "owner/name", "--state", str(path)], - now=utc(2026, 9, 29, 19, 31), - load=load, - ) - return status, printed.getvalue(), load_state(path) + self, load: default_branch.Loader, remote: FakeRemote + ) -> tuple[int, str]: + printed = io.StringIO() + with contextlib.redirect_stdout(printed): + status = clock.main( + ["--repo", REPO, "--send"], + now=utc(2026, 9, 29, 19, 31), + load=load, + remote=remote, + ) + return status, printed.getvalue() - def test_a_tick_that_reads_nothing_fails_and_keeps_the_state(self) -> None: + def test_a_tick_that_reads_nothing_fails_and_sends_nothing(self) -> None: def unreachable(_repo: str) -> tuple[dict[str, object], dict[str, str]]: raise GitHubError("gh: Server Error (HTTP 502)") - state = {DAILY_KEY: Handled(utc(2026, 9, 28, 19, 30), True)} - status, printed, after = self.run_main(unreachable, state) + remote = FakeRemote(marked(DAILY, utc(2026, 9, 28, 19, 30))) + status, printed = self.run_main(unreachable, remote) self.assertEqual(status, 1) self.assertIn("HTTP 502", printed) - self.assertEqual(after, state) + self.assertEqual(remote.dispatched, []) def test_an_unreadable_file_fails_the_tick_but_not_the_rest(self) -> None: - # The readable workflow is still delivered (in shadow mode, logged); - # the unreadable one is named, and the tick goes red over it. def load(_repo: str) -> tuple[dict[str, object], dict[str, str]]: return parsed({"backup.yml": ENROLLED}), {"broken.yml": "bad YAML"} - state = {DAILY_KEY: Handled(utc(2026, 9, 28, 19, 30), True)} - status, printed, after = self.run_main(load, state) + remote = FakeRemote(marked(DAILY, utc(2026, 9, 28, 19, 30))) + status, printed = self.run_main(load, remote) self.assertEqual(status, 1) self.assertIn("broken.yml: unreadable: bad YAML", printed) - self.assertIn("would send", printed) - self.assertEqual(after[DAILY_KEY].slot, utc(2026, 9, 29, 19, 30)) - - -class StateFileTest(unittest.TestCase): - def test_round_trip(self) -> None: - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "nested" / "state.json" - state = { - "backup.yml 30 19 * * *": Handled(utc(2026, 9, 28, 19, 30), True), - "x.yml 0 1 * * *": Handled(utc(2026, 9, 28, 1, 0), False), - } - save_state(path, state) - self.assertEqual(load_state(path), state) - - def test_missing_or_damaged_state_reads_as_empty(self) -> None: - # Empty state re-baselines every line, which sends nothing - the safe - # reading of a state that cannot be trusted. - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "state.json" - self.assertEqual(load_state(path), {}) - for damaged in ( - "{not json", - '["k"]', - '{"k": 1}', - '{"k": "2026-09-28T19:30Z"}', - '{"k": {"slot": "2026-09-28T19:30", "enrolled": true}}', - '{"k": {"slot": "2026-09-28T19:30Z", "enrolled": "yes"}}', - ): - with self.subTest(damaged=damaged): - path.write_text(damaged, encoding="utf-8") - with contextlib.redirect_stdout(io.StringIO()): - self.assertEqual(load_state(path), {}) + self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 29, 19, 30))]) if __name__ == "__main__": diff --git a/tests/test_default_branch.py b/tests/test_default_branch.py new file mode 100644 index 0000000..a7e217a --- /dev/null +++ b/tests/test_default_branch.py @@ -0,0 +1,229 @@ +"""Unit tests for reading the default branch's workflow files through a cache. + +The risks: an idle tick that downloads or parses anything, which is the load +this reader exists to avoid; a cached text or parse served for content it does +not belong to, which would deliver from a stale or tampered file; a cache whose +loss or corruption changes what the clock decides, which would make it state +rather than a cache; and a tree read in two queries that describes two +different trees. GitHub's GraphQL answers are replaced by canned ones in the +shape CI's live check exercises; parsing runs the real yq. +""" + +import os +import tempfile +import unittest +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from github_cron_trigger import default_branch, yq +from github_cron_trigger.default_branch import Cache, Entry, git_blob_id +from github_cron_trigger.github import GitHubError +from github_cron_trigger.workflow_triggers import schedule_entries, triggers + +ENROLLED = ( + 'on:\n schedule:\n - cron: "30 19 * * *"\n' + " repository_dispatch:\n types: [github-cron-trigger/backup.yml]\njobs: {}\n" +) +PUSH = "on: push\njobs: {}\n" +TREE = "1" * 40 +NEW_TREE = "2" * 40 + + +def oid(text: str) -> str: + return git_blob_id(text.encode("utf-8")) + + +class FakeGitHub: + """Answers the tree id from `tree`, a tree listing from `trees`, and blob + queries from `texts`, recording every call by kind.""" + + def __init__( + self, tree: str | None, trees: Mapping[str, Mapping[str, str]] + ) -> None: + self.tree = tree + self.trees = trees + self.texts = { + oid(text): text for files in trees.values() for text in files.values() + } + self.calls: list[str] = [] + self.fetched: list[str] = [] + + def __call__(self, query: str, variables: Mapping[str, str]) -> Any: + if "HEAD:.github/workflows" in query: + self.calls.append("tree") + return { + "repository": { + "object": None if self.tree is None else {"oid": self.tree} + } + } + if "$tree" in query: + self.calls.append("entries") + files = self.trees[variables["tree"]] + entries = [ + {"name": n, "type": "blob", "oid": oid(t)} for n, t in files.items() + ] + entries.append({"name": "README.md", "type": "blob", "oid": oid("readme")}) + entries.append({"name": "nested", "type": "tree", "oid": "3" * 40}) + return {"repository": {"object": {"entries": entries}}} + self.calls.append("texts") + answers: dict[str, Any] = {} + for index, part in enumerate(query.split('object(oid: "')[1:]): + asked = part[:40] + self.fetched.append(asked) + text = self.texts.get(asked) + answers[f"b{index}"] = { + "text": text, + "isBinary": False, + "isTruncated": False, + } + return {"repository": answers} + + +class LoadTest(unittest.TestCase): + def setUp(self) -> None: + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.parses = 0 + + def executable(self) -> str: + self.parses += 1 + return yq.executable() + + def load( + self, github: FakeGitHub, root: Path | None = None + ) -> tuple[dict[str, object], dict[str, str]]: + return default_branch.load( + "owner/name", Cache(root or self.root), github, self.executable, lambda: 0.0 + ) + + def test_a_first_read_reads_and_parses_every_workflow_file(self) -> None: + github = FakeGitHub(TREE, {TREE: {"backup.yml": ENROLLED, "ci.yml": PUSH}}) + documents, unreadable = self.load(github) + self.assertEqual(sorted(documents), ["backup.yml", "ci.yml"]) + self.assertEqual(unreadable, {}) + self.assertEqual(github.calls, ["tree", "entries", "texts"]) + self.assertEqual(sorted(github.fetched), sorted([oid(ENROLLED), oid(PUSH)])) + + def test_an_idle_read_asks_for_the_tree_id_and_nothing_else(self) -> None: + github = FakeGitHub(TREE, {TREE: {"backup.yml": ENROLLED, "ci.yml": PUSH}}) + first = self.load(github) + github.calls.clear() + self.parses = 0 + self.assertEqual(self.load(github), first) + self.assertEqual(github.calls, ["tree"]) + # Not even yq's version is checked: no process starts on an idle tick. + self.assertEqual(self.parses, 0) + + def test_a_changed_file_is_the_only_one_fetched_and_parsed(self) -> None: + changed = ENROLLED.replace("19", "20") + github = FakeGitHub( + TREE, + { + TREE: {"backup.yml": ENROLLED, "ci.yml": PUSH}, + NEW_TREE: {"backup.yml": changed, "ci.yml": PUSH}, + }, + ) + self.load(github) + github.tree = NEW_TREE + github.calls.clear() + github.fetched.clear() + documents, _ = self.load(github) + self.assertEqual(github.calls, ["tree", "entries", "texts"]) + self.assertEqual(github.fetched, [oid(changed)]) + entries = schedule_entries( + "backup.yml", triggers(documents["backup.yml"]).get("schedule") + ) + self.assertEqual([entry.cron for entry in entries], ["30 20 * * *"]) + + def test_losing_the_cache_costs_a_full_read_and_changes_nothing(self) -> None: + github = FakeGitHub(TREE, {TREE: {"backup.yml": ENROLLED}}) + with_cache = self.load(github) + github.calls.clear() + self.assertEqual(self.load(github, root=self.root / "elsewhere"), with_cache) + self.assertEqual(github.calls, ["tree", "entries", "texts"]) + + def test_no_workflow_directory_reads_nothing(self) -> None: + github = FakeGitHub(None, {}) + self.assertEqual(self.load(github), ({}, {})) + self.assertEqual(github.calls, ["tree"]) + + def test_a_file_that_does_not_parse_is_reported_every_time(self) -> None: + github = FakeGitHub(TREE, {TREE: {"broken.yml": "on: [push\n"}}) + for _ in range(2): + documents, unreadable = self.load(github) + self.assertEqual((documents, list(unreadable)), ({}, ["broken.yml"])) + + def test_an_answer_in_another_shape_is_a_github_error(self) -> None: + def odd(query: str, variables: Mapping[str, str]) -> Any: + return {"repository": {"object": {"id": "not an oid"}}} + + with self.assertRaises(GitHubError): + default_branch.load("owner/name", Cache(self.root), odd, self.executable) + + +class CacheTest(unittest.TestCase): + def setUp(self) -> None: + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.cache = Cache(Path(directory.name)) + + def test_git_blob_ids_match_git(self) -> None: + # The ids `git hash-object` gives these contents. + self.assertEqual(git_blob_id(b""), "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391") + self.assertEqual( + git_blob_id(b"hello\n"), "ce013625030ba8dba906f756967f9e9ca394464a" + ) + + def test_a_text_is_kept_only_under_its_own_id(self) -> None: + self.cache.put_text(oid(ENROLLED), ENROLLED) + self.assertEqual(self.cache.text(oid(ENROLLED)), ENROLLED) + self.cache.put_text(oid(PUSH), ENROLLED) + self.assertIsNone(self.cache.text(oid(PUSH))) + + def test_a_changed_cache_file_is_a_miss(self) -> None: + self.cache.put_text(oid(ENROLLED), ENROLLED) + assert self.cache.root is not None + stored = self.cache.root / "blobs" / oid(ENROLLED)[:2] / oid(ENROLLED) + stored.write_text(ENROLLED.replace("19", "20"), encoding="utf-8") + self.assertIsNone(self.cache.text(oid(ENROLLED))) + + def test_damaged_listings_and_parses_are_misses(self) -> None: + assert self.cache.root is not None + for kind in ("trees", f"parsed-{default_branch.PARSE_FORMAT}"): + path = self.cache.root / kind / TREE[:2] / TREE + path.parent.mkdir(parents=True) + path.write_text("{not json", encoding="utf-8") + self.assertIsNone(self.cache.entries(TREE)) + self.assertIsNone(self.cache.parsed(TREE)) + + def test_entries_round_trip(self) -> None: + entries = [Entry("a.yml", oid("a")), Entry("b.yaml", oid("b"))] + self.cache.put_entries(TREE, entries) + self.assertEqual(self.cache.entries(TREE), entries) + + def test_unused_entries_are_pruned_and_used_ones_kept(self) -> None: + self.cache.put_text(oid(ENROLLED), ENROLLED) + self.cache.put_text(oid(PUSH), PUSH) + assert self.cache.root is not None + unused = self.cache.root / "blobs" / oid(ENROLLED)[:2] / oid(ENROLLED) + os.utime(unused, (0, 0)) + # Pruned as of a moment just past the unused entry's maximum age; the + # other was written, so touched, just now. + self.cache.prune(default_branch.CACHE_MAX_AGE + 1) + self.assertFalse(unused.exists()) + self.assertEqual(self.cache.text(oid(PUSH)), PUSH) + + def test_no_directory_or_an_unwritable_one_only_misses(self) -> None: + none = Cache(None) + none.put_text(oid(ENROLLED), ENROLLED) + self.assertIsNone(none.text(oid(ENROLLED))) + with tempfile.NamedTemporaryFile() as blocker: + blocked = Cache(Path(blocker.name)) + blocked.put_text(oid(ENROLLED), ENROLLED) + self.assertIsNone(blocked.text(oid(ENROLLED))) + + +if __name__ == "__main__": + unittest.main()