From 1c8bff465bc9967cfbf2ff8b7f0b9de1f5be545c Mon Sep 17 00:00:00 2001 From: Jeffrey Wall <265409180+jeffwall-curlewlabs@users.noreply.github.com> Date: Mon, 5 Oct 2026 13:12:56 -0700 Subject: [PATCH 1/2] Keep the clock's state in the served repository The clock remembered what it had sent in a local state file, which tied delivery to one host: while that host slept every slot waited for GitHub's backstop, and a second host would have sent every slot again. A clock now claims a slot before sending it, by creating a ref under refs/github-cron-trigger-clock/ in the served repository. Ref creation is atomic, so of any number of clocks ticking at once exactly one sends each slot. A failed send releases its claim, so the next tick on any host retries. A line with no marks is baselined; a workflow that stops being enrolled loses its marks, so enrolling it again baselines it again; each line keeps its two newest marks. Marks of a workflow file that could not be read this tick are left alone. `--state` is gone: a host keeps nothing. Without `--send` a tick is a dry run that reports what it would do and writes nothing. The ledger's ref primitives (create with a note, delete, list) are shared with the marks. CI's live check runs two clocks at once against a workflow enrolled for it, clock-check.yml, and checks each slot is sent once. Closes #2. --- .github/workflows/ci.yml | 14 +- .github/workflows/clock-check.yml | 20 ++ README.md | 86 ++++-- SECURITY.md | 9 +- pyproject.toml | 2 +- src/github_cron_trigger/clock.py | 320 ++++++++++++--------- src/github_cron_trigger/clock_marks.py | 112 ++++++++ src/github_cron_trigger/slot_ledger.py | 73 +++-- tests/live_check.py | 129 ++++++++- tests/test_clock.py | 380 ++++++++++++++++--------- 10 files changed, 806 insertions(+), 339 deletions(-) create mode 100644 .github/workflows/clock-check.yml create mode 100644 src/github_cron_trigger/clock_marks.py 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..49d12d6 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,75 @@ 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. +- **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 +306,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 +406,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..e5d856e 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 .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 @@ -346,6 +381,7 @@ def main( argv: Sequence[str], now: datetime | None = None, load: workflow_files.Loader = workflow_files.load_default_branch, + 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/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/tests/live_check.py b/tests/live_check.py index ae1808f..189b045 100644 --- a/tests/live_check.py +++ b/tests/live_check.py @@ -24,21 +24,29 @@ 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 datetime import datetime, timedelta from pathlib import Path from github_cron_trigger import clock, freshness, slot_ledger, steps, workflow_files +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 @@ -182,14 +190,6 @@ def readers(repo: str, workflow_ref: str, workflow_sha: str) -> None: status == 1 and not outputs and "is not one of" in printed, "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) - _expect( - status == 0 and bool(handled), - f"a shadow-mode tick reads the default branch and baselines {len(handled)} line(s)", - ) status = freshness.main(["--repo", repo]) _expect( status & freshness.EXIT_INCOMPLETE == 0, @@ -197,6 +197,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 +302,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..4219c2b 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.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: workflow_files.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__": From aad28c00a86f440820f8414b24d2ef5ce34a1ece Mon Sep 17 00:00:00 2001 From: Jeffrey Wall <265409180+jeffwall-curlewlabs@users.noreply.github.com> Date: Mon, 5 Oct 2026 13:27:03 -0700 Subject: [PATCH 2/2] Make an idle tick ask only whether anything changed Every tick downloaded the full text of every workflow file on the default branch and parsed each with yq, though workflow files change a few times a day and a clock ticks every few minutes. For a repository with 52 workflow files that is about 1.3 MB and 52 yq runs per tick. A tick now asks for the id of the default branch's `.github/workflows` tree. When the tree is one it has read before, everything comes from a cache keyed by content id - the tree's entries, each file's parse - and nothing else is fetched or parsed. When it changed, only the files whose blob is new are fetched, in one batched query, and parsed. Cached texts are checked against their git blob id on every read, and anything unreadable is a miss, so the cache can never be stale and deleting it changes no decision. Entries unused for 30 days are pruned. Measured on that 52-file repository: a cold read is 4 queries, 1.3 MB and 3.1 s; an idle read is 1 query, 79 bytes, 0.6 s and no yq. Closes #4. --- README.md | 7 + src/github_cron_trigger/clock.py | 4 +- src/github_cron_trigger/default_branch.py | 323 ++++++++++++++++++++++ src/github_cron_trigger/freshness.py | 4 +- src/github_cron_trigger/workflow_files.py | 87 +----- tests/live_check.py | 35 ++- tests/test_clock.py | 4 +- tests/test_default_branch.py | 229 +++++++++++++++ 8 files changed, 604 insertions(+), 89 deletions(-) create mode 100644 src/github_cron_trigger/default_branch.py create mode 100644 tests/test_default_branch.py diff --git a/README.md b/README.md index 49d12d6..6bab4f3 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,13 @@ each line's two newest marks and removes the rest. - **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. diff --git a/src/github_cron_trigger/clock.py b/src/github_cron_trigger/clock.py index e5d856e..02112f2 100644 --- a/src/github_cron_trigger/clock.py +++ b/src/github_cron_trigger/clock.py @@ -43,7 +43,7 @@ from datetime import datetime 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, @@ -380,7 +380,7 @@ 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) 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/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 189b045..01b1b2b 100644 --- a/tests/live_check.py +++ b/tests/live_check.py @@ -42,10 +42,20 @@ import subprocess import sys import tempfile +from collections.abc import Mapping from datetime import datetime, timedelta from pathlib import Path - -from github_cron_trigger import clock, freshness, slot_ledger, steps, workflow_files +from typing import Any + +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 @@ -190,6 +200,27 @@ def readers(repo: str, workflow_ref: str, workflow_sha: str) -> None: status == 1 and not outputs and "is not one of" in printed, "a dispatch of a line this workflow does not carry is refused", ) + with tempfile.TemporaryDirectory() as tmp: + 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( + 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( status & freshness.EXIT_INCOMPLETE == 0, diff --git a/tests/test_clock.py b/tests/test_clock.py index 4219c2b..4e61c9d 100644 --- a/tests/test_clock.py +++ b/tests/test_clock.py @@ -23,7 +23,7 @@ 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 ( ScheduledWorkflow, plan, @@ -490,7 +490,7 @@ class MainTest(unittest.TestCase): runs the clock reports it; a file it could not read must not stop the rest.""" def run_main( - self, load: workflow_files.Loader, remote: FakeRemote + self, load: default_branch.Loader, remote: FakeRemote ) -> tuple[int, str]: printed = io.StringIO() with contextlib.redirect_stdout(printed): 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()