From 7dff4d1ad2d8eaf0dbcc359a1043c2351f364914 Mon Sep 17 00:00:00 2001 From: Jeffrey Wall <265409180+jeffwall-curlewlabs@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:36:29 -0700 Subject: [PATCH] Never leave a claimed slot marked and unsent A clock claims a slot's mark, then reads the ledger, then sends. Two paths left the mark in place with nothing sent, so every later tick read the slot as sent and only GitHub's backstop delivered it, late: - The ledger read after the claim failed. It now releases the mark, as a failed send already did. - The claim's create applied but GitHub answered otherwise. gh_api retries a POST whose answer was lost, and the retry is told the ref exists, so the clock took its own mark for another clock's; or the lost answer was the last attempt's, and the claim failed with the mark in place. Each claim's note now carries a random id, so its blob is that claim's alone, and a claim that is refused or fails reads the mark and holds it when the mark points at its own blob. The live check claims the same mark twice with one id, which is what a retried create that applied looks like, and confirms the claim holds it: that rests on the note hashing to the blob GitHub made. --- README.md | 11 +-- pyproject.toml | 2 +- src/github_cron_trigger/clock.py | 30 +++++---- src/github_cron_trigger/clock_marks.py | 67 +++++++++++++++---- src/github_cron_trigger/slot_ledger.py | 16 ++++- tests/live_check.py | 25 +++++-- tests/test_clock.py | 92 ++++++++++++++++++++++++-- 7 files changed, 196 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index 6bab4f3..ec2b0a5 100644 --- a/README.md +++ b/README.md @@ -251,9 +251,12 @@ create succeeds, and only that clock sends. So: 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. +- **A failed send releases its claim**, as does a failed ledger read before it, + so the next tick, on any host, sends the slot again. A claim that GitHub + applied but answered as failed or refused is still recognized as the clock's + own: each claim's ref points at a note unique to it. 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 @@ -289,7 +292,7 @@ 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.2.0" +pip install "github-cron-trigger @ git+https://github.com/curlewlabs-com/github-cron-trigger@v0.2.1" ``` Or check out a release tag and run it as `PYTHONPATH=src python3 -m diff --git a/pyproject.toml b/pyproject.toml index 57a9ae5..bc054a0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi" [project] name = "github-cron-trigger" -version = "0.2.0" +version = "0.2.1" 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 02112f2..6a5c6e4 100644 --- a/src/github_cron_trigger/clock.py +++ b/src/github_cron_trigger/clock.py @@ -323,26 +323,28 @@ def tick( print(f"clock: {what}: {verb} (dry run)") continue try: - 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") - continue + claimed = remote.claim(repo, mark) except (GitHubError, ValueError) as exc: problems.append(f"{what}: {exc}") continue + if not claimed: + 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 try: + if remote.is_done(repo, workflow_file, step.schedule, step.slot): + print(f"clock: {what}: already recorded done; not sent") + continue 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. + # The ledger read or the send failed, so the mark comes off again + # and the next tick on any host tries the slot; left on, it would + # read as sent. A ValueError is a name no retry fixes; it is + # reported on every tick. problems.append(f"{what}: {exc}") try: remote.release(repo, mark) diff --git a/src/github_cron_trigger/clock_marks.py b/src/github_cron_trigger/clock_marks.py index 6649ee9..bf86a09 100644 --- a/src/github_cron_trigger/clock_marks.py +++ b/src/github_cron_trigger/clock_marks.py @@ -5,7 +5,8 @@ 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. +which slot it marks and which claim made it. Only the claim that made a mark +reads what it points at (GitHubMarks.claim). 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 @@ -16,10 +17,14 @@ 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. +one clock however many tick at once. A send that fails, or the ledger read +before it, releases its mark, so the next tick on any host tries the slot +again. A create can apply and still answer otherwise - a retried POST is told +its own ref exists, or the answer to the one that applied is lost - so a claim +that is refused or fails reads the mark, and holds it when the mark points at +its own blob. A clock stopped between its claim and its send, or one that can +neither create nor read the mark, 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 @@ -28,12 +33,15 @@ """ import re -from collections.abc import Iterable +import secrets +from collections.abc import Callable, 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 +from .default_branch import git_blob_id +from .github import GitHubError +from .slot_ledger import create_ref, cron_segment, delete_ref, list_refs, ref_target MARK_NAMESPACE = "github-cron-trigger-clock" @@ -85,14 +93,34 @@ def parse_mark(name: str) -> Mark | 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" +def note(mark: Mark, claim_id: str) -> str: + """The text of a mark's blob. It names the slot, and the claim by a random + id, so its blob is that claim's alone; nothing about the host, since a + public repository's refs are readable by anyone.""" + return ( + f"github-cron-trigger clock: claimed {mark.workflow_file}" + f" slot {format_slot(mark.slot)} (claim {claim_id})\n" + ) + + +def new_claim_id() -> str: + return secrets.token_hex(16) class GitHubMarks: - """Marks read and written through the GitHub API.""" + """Marks read and written through the GitHub API. The ref calls and the + claim id are parameters so a test can replay a create that applied but + answered otherwise.""" + + def __init__( + self, + create: Callable[[str, str, str], bool] = create_ref, + target: Callable[[str, str], str | None] = ref_target, + claim_id: Callable[[], str] = new_claim_id, + ) -> None: + self._create = create + self._target = target + self._claim_id = claim_id def read(self, repo: str) -> list[Mark]: """Every mark the repository holds. A ref in the namespace that is not @@ -101,8 +129,19 @@ def read(self, repo: str) -> list[Mark]: 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)) + """Create the mark; True when this claim holds it, False when another + does. A create that fails without applying raises.""" + text = note(mark, self._claim_id()) + own = git_blob_id(text.encode("utf-8")) + try: + if self._create(repo, mark.path, text): + return True + except GitHubError: + target = self._target(repo, mark.path) + if target is None: + raise + return target == own + return self._target(repo, mark.path) == own def release(self, repo: str, mark: Mark) -> None: delete_ref(repo, mark.path) diff --git a/src/github_cron_trigger/slot_ledger.py b/src/github_cron_trigger/slot_ledger.py index 2e8aef5..3a1df05 100644 --- a/src/github_cron_trigger/slot_ledger.py +++ b/src/github_cron_trigger/slot_ledger.py @@ -98,15 +98,25 @@ def is_done( ) -> bool: """Whether the ledger records this slot (and part) as done.""" path = ref_path(workflow_file, schedule, slot, part) + return ref_target(repo, path) is not None + + +def ref_target(repo: str, path: str) -> str | None: + """The object `refs/` points at, or None when there is no such ref.""" try: # git/ref/ (singular) matches exactly; git/matching-refs/ would also # answer for a longer ref that merely starts with this one. - gh_api([f"repos/{check_repo(repo)}/git/ref/{path}"]) + out = gh_api( + [f"repos/{check_repo(repo)}/git/ref/{path}", "--jq", ".object.sha"] + ) except HttpError as exc: if exc.status == 404: - return False + return None raise - return True + target = out.strip() + if not _OBJECT.match(target): + raise GitHubError(f"ref {path} answered {target!r}, not an object name") + return target def record_note(sha: str) -> str: diff --git a/tests/live_check.py b/tests/live_check.py index 01b1b2b..62e0238 100644 --- a/tests/live_check.py +++ b/tests/live_check.py @@ -31,8 +31,11 @@ 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. +an older slot marked, exactly one of them sends the newest. Marking that older +slot checks the claim itself: a claim made twice holds the mark it made, as a +retried create that applied must, and a different claim of it does not. The +marks live in the repository (clock_marks.py), and are removed again whatever +the outcome. """ import argparse @@ -311,11 +314,19 @@ def clocks(repo: str, required: bool) -> None: "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)) - ), + older = mark_for( + CLOCK_CHECK, YEARLY, latest_slot(YEARLY, newest - timedelta(minutes=1)) + ) + # The same claim made twice is what a retried create that applied looks + # like; holding the mark rests on its note hashing to the blob GitHub made. + retried = GitHubMarks(claim_id=lambda: f"live-check-{os.getpid()}") + _expect( + retried.claim(repo, older) and retried.claim(repo, older), + "a claim refused by its own earlier create holds the mark", + ) + _expect( + not MARKS.claim(repo, older), + "a claim of a mark another claim made is not held", ) outputs = _concurrent_ticks(repo) _expect( diff --git a/tests/test_clock.py b/tests/test_clock.py index 4e61c9d..c3ae859 100644 --- a/tests/test_clock.py +++ b/tests/test_clock.py @@ -6,8 +6,10 @@ 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; 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 +failed send or ledger read that leaves its slot marked, so no clock retries it; +a claim GitHub applied but answered as failed or refused, read as another +clock's, which strands the slot the same way; 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 @@ -18,7 +20,7 @@ import contextlib import io import unittest -from collections.abc import Iterable +from collections.abc import Callable, Iterable from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo @@ -31,13 +33,14 @@ stale_marks, tick, ) -from github_cron_trigger.clock_marks import Mark, mark_for +from github_cron_trigger.clock_marks import GitHubMarks, Mark, mark_for from github_cron_trigger.cron_slots import ( UTC, CronSchedule, UnsupportedCron, parse_cron, ) +from github_cron_trigger.default_branch import git_blob_id from github_cron_trigger.github import GitHubError from github_cron_trigger.workflow_triggers import MisconfiguredWorkflow @@ -326,10 +329,12 @@ def __init__( marks: Iterable[Mark] = (), done: Iterable[datetime] = (), failing_dispatches: int = 0, + failing_done_reads: int = 0, ) -> None: self.marks = set(marks) self.done = set(done) self.failing_dispatches = failing_dispatches + self.failing_done_reads = failing_done_reads self.dispatched: list[tuple[str, datetime]] = [] self.snapshot: list[Mark] | None = None @@ -353,6 +358,9 @@ def remove(self, repo: str, marks: Iterable[Mark]) -> None: def is_done( self, repo: str, workflow_file: str, schedule: CronSchedule, slot: datetime ) -> bool: + if self.failing_done_reads: + self.failing_done_reads -= 1 + raise GitHubError("gh: Server Error (HTTP 502)") return slot in self.done def dispatch( @@ -430,6 +438,22 @@ def test_a_failed_send_releases_its_mark_so_the_next_tick_retries(self) -> None: self.assertEqual(problems, []) self.assertEqual(remote.dispatched, [("backup.yml", utc(2026, 9, 28, 19, 30))]) + def test_a_failed_ledger_read_releases_its_mark_so_the_next_tick_retries( + self, + ) -> None: + remote = FakeRemote( + marked(DAILY, utc(2026, 9, 27, 19, 30)), failing_done_reads=1 + ) + problems, _ = run_tick(remote, self.enrolled, utc(2026, 9, 28, 19, 31)) + self.assertEqual(len(problems), 1) + self.assertEqual(remote.dispatched, []) + self.assertNotIn( + mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30)), remote.marks + ) + 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)] @@ -485,6 +509,66 @@ def read_marks(self, repo: str) -> list[Mark]: self.assertEqual((remote.marks, remote.dispatched), (set(), [])) +class ClaimTest(unittest.TestCase): + """GitHub can apply a ref create and still answer otherwise: gh_api retries + a POST whose answer was lost, and the retry is told the ref exists; or the + lost answer is the last attempt's, and the call fails. Read as another + clock's claim, or as no claim, the mark would stay with nothing sent. The + refs here are a dictionary of path to blob id; the blob ids are git's.""" + + MARK = mark_for("backup.yml", LINE, utc(2026, 9, 28, 19, 30)) + + def setUp(self) -> None: + self.refs: dict[str, str] = {} + + def create(self, repo: str, path: str, text: str) -> bool: + if path in self.refs: + return False + self.refs[path] = git_blob_id(text.encode("utf-8")) + return True + + def marks( + self, create: Callable[[str, str, str], bool], claim_id: str + ) -> GitHubMarks: + return GitHubMarks( + create=create, + target=lambda repo, path: self.refs.get(path), + claim_id=lambda: claim_id, + ) + + def test_a_retry_refused_by_its_own_create_holds_the_mark(self) -> None: + def retried(repo: str, path: str, text: str) -> bool: + self.create(repo, path, text) + return self.create(repo, path, text) + + self.assertTrue(self.marks(retried, "a").claim(REPO, self.MARK)) + + def test_a_create_that_applied_but_failed_holds_the_mark(self) -> None: + def lost(repo: str, path: str, text: str) -> bool: + self.create(repo, path, text) + raise GitHubError("`gh api` timed out") + + self.assertTrue(self.marks(lost, "a").claim(REPO, self.MARK)) + + def test_a_create_that_failed_without_applying_raises(self) -> None: + def failed(repo: str, path: str, text: str) -> bool: + raise GitHubError("gh: Server Error (HTTP 502)") + + with self.assertRaises(GitHubError): + self.marks(failed, "a").claim(REPO, self.MARK) + self.assertEqual(self.refs, {}) + + def test_another_claims_mark_is_not_held_however_the_create_answers( + self, + ) -> None: + def failed(repo: str, path: str, text: str) -> bool: + raise GitHubError("gh: Server Error (HTTP 502)") + + self.assertTrue(self.marks(self.create, "first").claim(REPO, self.MARK)) + self.assertFalse(self.marks(self.create, "second").claim(REPO, self.MARK)) + self.assertFalse(self.marks(failed, "third").claim(REPO, self.MARK)) + + class MainTest(unittest.TestCase): """A tick that could not read what it had to deliver must fail, so whatever runs the clock reports it; a file it could not read must not stop the rest."""