diff --git a/progress/apply.py b/progress/apply.py index 07c011d..2b2fbe3 100644 --- a/progress/apply.py +++ b/progress/apply.py @@ -22,6 +22,11 @@ "our": anyone may open a pull request on a `progress/*` branch, and branch names are a pure function of the window, so honouring a stranger's would let them decide what this operator is allowed to publish -- permanently, by opening and closing one pull request. + +Reconciling the window is not enough on its own, because an area can hold more than one open report +and the others are not reconciled by anything. Once ours is open we therefore close the area's +reports that can no longer merge -- see `superseded_prs` for the two ways that happens. Without it +the repository grows one unmergeable report per area per round and never sheds them. """ import json @@ -143,6 +148,142 @@ def existing_pr(branch, repo=gh.ROADMAP_REPO, owner=None, states=("merged",)): return rows[0] if rows else None +def report_meta(body): + """The `tauceti-progress-pr:v1` metadata a report carries in its body, or None. + + `pr_body` writes `from_sha`, `to_sha` and the window's PR numbers there, so a sibling states its + own window exactly rather than in the seven-character abbreviations of a branch name. That is + what makes a containment test possible without a TauCeti clone. + + The body is mutable and anyone may edit it, so nothing here is trusted for a decision that could + destroy work: the caller only ever uses it to prove that a report covers LESS than ours, and a + forged wider claim merely spares a report from being closed. + """ + marker = BODY_MARKER.split("{}")[0] + for line in (body or "").splitlines(): + line = line.strip() + if not line.startswith(marker): + continue + blob = line[len(marker):] + blob = blob[:blob.rfind("-->")] if "-->" in blob else blob + try: + meta = json.loads(blob) + except json.JSONDecodeError: + return None + return meta if isinstance(meta, dict) else None + return None + + +def superseded_prs(open_prs, area, live_cursor, plan_cursor, keep_branch, owners, our_prs=None): + """Open reports for `area` that this run's report replaces: `[(row, reason)]`. + + **Nothing is swept unless we are the live report.** `live_cursor` is read from the roadmap + repository now; `plan_cursor` is where the cursor was when this run planned. If they differ, a + sibling landed in between and it is OUR report that is dead, not anyone else's. Sweeping then + inverts the rule below and closes the one report that can still merge, so the answer is to sweep + nothing and let the next round reconcile. + + Given that, two things put a report beyond saving, and neither closes itself. + + **Its window no longer starts at the cursor.** The merge gate requires a byte-exact append at the + cursor recorded in `PROGRESS.md`, so a report whose `from_sha` is not that cursor can never + satisfy it -- dead by construction, not merely behind. This is what happens the moment a sibling + for the same area merges: the cursor advances and every other open report is orphaned instantly. + Nothing noticed, so they accumulated. + + **It starts at the cursor and we cover everything it covers.** Proved, not inferred: same + `from_sha`, and its window's PR numbers are a proper subset of ours. Creation order will not do + this job. A worker that planned early and stalled creates a NARROW report late, so "newest wins" + would close a wider report and lose coverage that had already been written; and neither a + timestamp nor a pull request number says anything about which window contains which. Containment + is antisymmetric, so two workers racing cannot each conclude the other is dominated, and the + stalled-worker case resolves correctly whichever of them opens first. + + A sibling whose window is incomparable to ours is left alone. So is every report, if `our_prs` is + unknown: closing then would assert a containment we have not established, and a duplicate left + for the next round is far cheaper than a report discarded. + + Scoped to reports we could have opened, like every other judgement in this module. Branch names + are a pure function of the window, so a stranger can create one; closing theirs would let this + operator silently veto someone else's contribution. The cost is real and worth naming: reports + published from another operator's fork are never swept, so this does not by itself eliminate a + multi-operator pile-up. + """ + if not live_cursor or not plan_cursor or live_cursor != plan_cursor: + return [] + ours = set(our_prs) if our_prs is not None else None + out = [] + for row in open_prs: + parts = (row.get("headRefName") or "").split("/") + if len(parts) != 3 or parts[-1] != area: + continue + if row.get("headRefName") == keep_branch: + continue + if ((row.get("headRepositoryOwner") or {}).get("login") or "") not in owners: + continue + from7 = parts[1].split("-")[0] if "-" in parts[1] else "" + if not from7: + continue + if not live_cursor.startswith(from7): + out.append((row, f"its window starts at {from7}, but the {area} cursor is now " + f"{live_cursor[:7]}, so it can never append")) + continue + if ours is None: + continue + meta = report_meta(row.get("body")) + if not meta or meta.get("from_sha") != live_cursor: + continue + try: + theirs = {int(n) for n in (meta.get("prs") or [])} + except (TypeError, ValueError): + continue + if theirs < ours: + out.append((row, f"its window covers {len(theirs)} of the {len(ours)} pull requests " + f"this one reports, from the same cursor {from7}")) + return out + + +def close_superseded(rows, replacement_url): + """Close each superseded report, saying what replaced it. Returns the number that failed. + + A failure here is never fatal -- this only runs once the replacement is open, so the window is + published either way -- but it is returned rather than swallowed, because the caller has to keep + retrying. `sweep` is what makes that retry actually happen. + """ + failed = 0 + for row, reason in rows: + note = f"Superseded by {replacement_url}: {reason}." + try: + gh.gh(["pr", "close", str(row["number"]), "--repo", gh.ROADMAP_REPO, + "--comment", note]) + print(f"closed superseded #{row['number']}: {reason}") + except gh.GhError as exc: + failed += 1 + print(f"could not close superseded #{row['number']} ({exc}); leaving it open") + return failed + + +def sweep(plan, keep_branch, replacement_url): + """Close the area's stranded reports. Safe to call whenever our report is open. + + Idempotent and deliberately reachable from BOTH paths in `run` -- the one that has just opened + the report and the one that finds it already open. That is not tidiness: a sweep that only ran + on the create path would never be retried at all, because the next run for the same window stops + at the in-flight check long before reaching it. A cleanup outage would then leave the pile-up + forever while every run reported success. + + Only `gh` failures are caught. A parsing or programming error here is a bug and must surface. + """ + live = files.cursor(gh.file_on_default_branch(plan["progress_path"]) or "") or "" + rows = superseded_prs( + gh.open_progress_prs(), plan["roadmap"], live, plan["from_sha"], keep_branch, + {gh.ROADMAP_REPO.split("/")[0], _own_login()}, our_prs=plan.get("prs"), + ) + if not rows: + return 0 + return close_superseded(rows, replacement_url) + + def remote_branch_exists(roadmap_dir, branch, remote="origin"): proc = _run(["git", "ls-remote", "--exit-code", "--heads", remote, branch], roadmap_dir, check=False) @@ -248,6 +389,12 @@ def run(plan, status_body_file, section_body_file, roadmap_dir, dry_run=False, v open_pr = own_pr(branch, states=("open",)) if open_pr is not None: print(f"already open, in flight: {open_pr['url']}") + # Sweep here too. This is the path every run after the first takes, so it is the only one + # that can retry a sweep an earlier run failed or never reached. + try: + sweep(plan, branch, open_pr.get("url") or "the open report") + except gh.GhError as exc: + print(f"could not sweep superseded reports ({exc}); the report is open regardless") return EX_NOPROGRESS # A CLOSED pull request means this window was refused, and reopening it every day is the loop @@ -337,4 +484,18 @@ def run(plan, status_body_file, section_body_file, roadmap_dir, dry_run=False, v "--title", title, "--body", body, ]) print(out.strip()) + + # Only now that ours is open: sweep the area's dead and superseded reports. Ordered this way so + # a failure here costs a tidy-up, never a publication -- and so we never close the only open + # report for an area. + # + # Without this the repository accumulates one unmergeable report per area per round, from two + # directions: a sibling merging orphans every other report for that area instantly, and any + # repository-wide breakage (a red `main` fails every report's `build`) makes each round open a + # fresh one that also cannot merge, since the planner's staleness expiry stops the previous one + # marking the area in flight. Neither ever closes itself. See `superseded_prs`. + try: + sweep(plan, branch, out.strip().splitlines()[-1] if out.strip() else "the new report") + except gh.GhError as exc: + print(f"could not sweep superseded reports ({exc}); the new report is open regardless") return 0 diff --git a/progress/gh.py b/progress/gh.py index 58eab1c..0c2a33f 100644 --- a/progress/gh.py +++ b/progress/gh.py @@ -13,6 +13,7 @@ right now". """ +import base64 import json import subprocess import time @@ -140,7 +141,24 @@ def open_progress_prs(repo=ROADMAP_REPO, branch_prefix="progress/"): out = gh([ "pr", "list", "--repo", repo, "--state", "open", "--limit", "200", "--json", - "number,headRefName,title,url,createdAt,headRepositoryOwner", + "number,headRefName,title,url,createdAt,headRepositoryOwner,body", ]) rows = json.loads(out) return [r for r in rows if (r.get("headRefName") or "").startswith(branch_prefix)] + + +def file_on_default_branch(path, repo=ROADMAP_REPO, ref="main"): + """The text of `path` on `ref`, or None if it is not there. + + Read through the API rather than from a clone on purpose. A worker's checkout is a snapshot from + whenever it started, and the questions this answers -- where is the cursor NOW, is this report + still the live one -- are exactly the ones a stale snapshot gets wrong. + """ + try: + raw = gh(["api", f"repos/{repo}/contents/{path}?ref={ref}", "--jq", ".content"]) + except GhError: + return None + try: + return base64.b64decode("".join(raw.split())).decode("utf-8", "surrogateescape") + except (ValueError, UnicodeDecodeError): + return None diff --git a/tests/test_supersede.py b/tests/test_supersede.py new file mode 100644 index 0000000..3ca55e4 --- /dev/null +++ b/tests/test_supersede.py @@ -0,0 +1,201 @@ +"""Tests for superseding: an area's unmergeable reports are closed once a live one is open. + +Two things strand a report, and before this neither closed itself, so they accumulated one per area +per round: a sibling merging moves the cursor and orphans every other open report for that area, and +a repository-wide breakage makes every round open a fresh report that also cannot merge. +""" + +import json +import pathlib +import sys + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from progress import apply # noqa: E402 + +failures = [] + + +def check(name, fn): + try: + fn() + except Exception as exc: # noqa: BLE001 + failures.append(name) + print(f"FAIL {name}: {type(exc).__name__}: {exc}") + else: + print(f"ok {name}") + + +OURS = "TauCetiProject" +OWNERS = {OURS} +CURSOR = "37aec57229a4a5828884027165b25804aac01ac8" + + +OLD_CURSOR = "b21862652ed4e0e0cb29351a0e78338370755d0a" +OUR_PRS = [6328, 6334, 6374, 6409] + + +def body(from_sha=CURSOR, to_sha="787733a", prs=(6328, 6334)): + meta = {"roadmap": "ModularForms", "from_sha": from_sha, "to_sha": to_sha, + "prs": sorted(prs), "version": "v1"} + return apply.BODY_MARKER.replace("{}", json.dumps(meta, sort_keys=True, + separators=(",", ":"))) + "\nprose\n" + + +def pr(number, area="ModularForms", from7="37aec57", to7="787733a", owner=OURS, body_text=None, + prs=(6328, 6334), meta_from=None): + return {"number": number, "url": f"https://example.invalid/{number}", + "headRefName": f"progress/{from7}-{to7}/{area}", + "headRepositoryOwner": {"login": owner}, + "body": body_text if body_text is not None + else body(from_sha=meta_from or CURSOR, to_sha=to7, prs=prs)} + + +def branch(from7="37aec57", to7="28e93d9", area="ModularForms"): + return f"progress/{from7}-{to7}/{area}" + + +def sweep(rows, live=CURSOR, planned=CURSOR, keep=None, owners=OWNERS, our_prs=OUR_PRS): + return apply.superseded_prs(rows, "ModularForms", live, planned, keep or branch(), owners, + our_prs=our_prs) + + +def test_an_orphan_whose_window_predates_the_cursor_is_closed(): + rows = sweep([pr(372, from7="b218626", to7="0038168", meta_from=OLD_CURSOR)]) + assert [r["number"] for r, _ in rows] == [372] + assert "can never append" in rows[0][1] + + +def test_a_dominated_report_at_the_same_cursor_is_closed(): + rows = sweep([pr(394, prs=(6328, 6334))]) + assert [r["number"] for r, _ in rows] == [394] + assert "covers 2 of the 4" in rows[0][1] + + +def test_a_stale_worker_sweeps_nothing(): + """Codex's case. We planned at C0; a sibling landed and the cursor is now C1, so OUR report is + the dead one. Sweeping on the stale cursor would close the only report that can still merge.""" + live_report = pr(399, from7="37aec57", to7="28e93d9") + assert sweep([live_report], live=CURSOR, planned=OLD_CURSOR) == [] + + +def test_a_wider_report_is_never_closed_by_a_narrower_one(): + """A worker that planned early and stalled opens a NARROW report late. Creation order would + close the wider one and lose coverage already written; containment does not.""" + wider = pr(410, prs=(6328, 6334, 6374, 6409, 6444)) + assert sweep([wider], our_prs=[6328, 6334]) == [] + + +def test_two_racing_workers_cannot_close_each_other(): + a = pr(410, to7="aaaaaaa", prs=(6328, 6334)) + b = pr(411, to7="bbbbbbb", prs=(6328, 6334, 6374, 6409)) + a_closes = sweep([b], keep=branch(to7="aaaaaaa"), our_prs=[6328, 6334]) + b_closes = sweep([a], keep=branch(to7="bbbbbbb"), our_prs=[6328, 6334, 6374, 6409]) + assert [r["number"] for r, _ in a_closes] == [] + assert [r["number"] for r, _ in b_closes] == [410] + + +def test_an_incomparable_window_is_kept(): + """Neither contains the other, so neither supersedes the other.""" + assert sweep([pr(410, prs=(6328, 9999))]) == [] + + +def test_an_identical_window_is_kept(): + """A proper subset, not any subset: an exact duplicate is not dominated, and closing on equality + would let two workers close each other again.""" + assert sweep([pr(410, prs=OUR_PRS)]) == [] + + +def test_unknown_own_window_closes_no_same_cursor_report(): + assert sweep([pr(394, prs=(6328,))], our_prs=None) == [] + + +def test_unknown_own_window_still_closes_orphans(): + """Being dead is a property of the report alone.""" + rows = sweep([pr(372, from7="b218626", to7="0038168", meta_from=OLD_CURSOR)], our_prs=None) + assert [r["number"] for r, _ in rows] == [372] + + +def test_a_report_without_readable_metadata_is_kept(): + assert sweep([pr(394, body_text="no marker here")]) == [] + assert sweep([pr(394, body_text=apply.BODY_MARKER.replace("{}", "not json"))]) == [] + + +def test_metadata_claiming_a_different_cursor_is_kept(): + """The body is mutable, so it is only ever used to prove a report covers LESS.""" + assert sweep([pr(394, meta_from=OLD_CURSOR, prs=(6328,))]) == [] + + +def test_our_own_branch_is_never_closed(): + assert sweep([pr(399, to7="28e93d9")], keep=branch(to7="28e93d9")) == [] + + +def test_another_area_is_untouched(): + assert sweep([pr(396, area="ArithmeticDirichletSeries", from7="8745177")]) == [] + + +def test_a_strangers_report_is_never_closed(): + assert sweep([pr(500, from7="b218626", owner="a-stranger", meta_from=OLD_CURSOR)]) == [] + + +def test_a_malformed_branch_is_ignored(): + bad = {"number": 9, "headRefName": "progress/ModularForms", + "headRepositoryOwner": {"login": OURS}, "body": body()} + worse = {"number": 10, "headRefName": "progress/nowindow/ModularForms", + "headRepositoryOwner": {"login": OURS}, "body": body()} + assert sweep([bad, worse]) == [] + + +def test_the_whole_modular_forms_pileup_is_swept(): + rows = sweep( + [pr(372, from7="b218626", to7="0038168", meta_from=OLD_CURSOR), + pr(381, from7="b218626", to7="0430506", meta_from=OLD_CURSOR), + pr(394, prs=(6328, 6334)), + pr(399, to7="28e93d9", prs=OUR_PRS)], + keep=branch(to7="28e93d9")) + assert sorted(r["number"] for r, _ in rows) == [372, 381, 394] + + +def test_report_meta_reads_the_window(): + meta = apply.report_meta(body(prs=(1, 2, 3))) + assert meta["from_sha"] == CURSOR and meta["prs"] == [1, 2, 3] + assert apply.report_meta("") is None + assert apply.report_meta(None) is None + + +def test_a_failed_close_is_counted_not_swallowed(): + """The caller has to know, so it keeps retrying on the in-flight path.""" + def boom(args, **kw): + raise apply.gh.GhError("gh exploded") + orig = apply.gh.gh + apply.gh.gh = boom + try: + assert apply.close_superseded([(pr(372), "because")], "https://example.invalid/399") == 1 + finally: + apply.gh.gh = orig + + +def test_a_programming_error_in_a_close_is_not_hidden(): + def boom(args, **kw): + raise TypeError("a real bug") + orig = apply.gh.gh + apply.gh.gh = boom + try: + apply.close_superseded([(pr(372), "because")], "url") + except TypeError: + pass + else: + raise AssertionError("a non-gh error must surface") + finally: + apply.gh.gh = orig + + +for _name, _fn in sorted(globals().items()): + if _name.startswith("test_") and callable(_fn): + check(_name, _fn) + +print() +if failures: + print(f"{len(failures)} failure(s): {', '.join(failures)}") + sys.exit(1) +print("all tests passed")