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] 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__":