From ff2c341b1512b850d77d76eaebbeea17823fb82c Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Fri, 14 Aug 2026 15:48:20 +0300 Subject: [PATCH 1/2] fix: merge test/test-full leaf variants per-cell instead of wholesale --- .../tests/therock_status_document_test.py | 116 ++++++++++++++++-- .../therock_status_document.py | 66 ++++++++++ .../therock_update_status_json.py | 19 ++- 3 files changed, 182 insertions(+), 19 deletions(-) diff --git a/scripts/receive_therock/tests/therock_status_document_test.py b/scripts/receive_therock/tests/therock_status_document_test.py index 58fe1e7afc..36abf609d3 100644 --- a/scripts/receive_therock/tests/therock_status_document_test.py +++ b/scripts/receive_therock/tests/therock_status_document_test.py @@ -27,6 +27,7 @@ StatusDocument, Variant, _merge_variant_leaf, + merge_matrix_test_leaf, ) @@ -592,14 +593,14 @@ def test_upsert_test_guard_is_per_arch() -> None: def test_upsert_test_replaces_whole_leaf_with_variants_atomically() -> None: # A pytorch.test arch leaf is ONE workflow run; its matrix cells are jobs in - # that run, carried on the leaf's `variants`. Unlike the build phase, test - # leaves are NOT merged cell-by-cell: a higher run_attempt replaces the - # entire leaf (and its full variant list) atomically. This is the - # rerun-of-failed-jobs case -- GitHub re-stamps every job with the new - # attempt, so the producer hands us a complete attempt-2 snapshot and the - # old attempt-1 variants are dropped wholesale, not preserved per-cell. - # Re-running failed jobs keeps the SAME run_id and only bumps run_attempt; - # the variants repeat the leaf's run_id. + # that run, carried on the leaf's `variants`. Like the build phase, test + # leaves with variants are merged cell-by-cell (see `merge_matrix_test_leaf`) + # rather than replaced wholesale. This is the rerun-of-failed-jobs case -- + # GitHub re-stamps every job with the new attempt, so every cell's higher + # run_attempt wins the per-cell guard and the net effect looks atomic: all + # attempt-1 values end up superseded. Re-running failed jobs keeps the + # SAME run_id and only bumps run_attempt; the variants repeat the leaf's + # run_id. run_id = 12345900 doc = StatusDocument() doc.upsert_leaf( @@ -662,9 +663,10 @@ def test_upsert_test_replaces_whole_leaf_with_variants_atomically() -> None: assert all(v.status is Status.success for v in leaf.variants) -def test_upsert_test_with_variants_rejects_lower_attempt_wholesale() -> None: - # The leaf-level guard protects the whole test leaf, variants included: a - # stale lower-attempt snapshot cannot clobber a newer one even cell-by-cell. +def test_upsert_test_with_variants_rejects_lower_attempt_per_cell() -> None: + # The variant path merges per-cell (like build) and always reports the + # write as accepted, but the per-cell guard still protects the actual + # data: a stale lower-attempt snapshot cannot clobber a newer cell. doc = StatusDocument() doc.upsert_leaf( "linux", @@ -677,7 +679,7 @@ def test_upsert_test_with_variants_rejects_lower_attempt_wholesale() -> None: variants=[_variant(matrix={"py": "3.11"}, run_attempt=2)], ), ) - assert not doc.upsert_leaf( + assert doc.upsert_leaf( "linux", "gfx942", "pytorch", @@ -689,11 +691,99 @@ def test_upsert_test_with_variants_rejects_lower_attempt_wholesale() -> None: ), ) leaf = doc.pipelines.pytorch.test["linux"]["gfx942"] - assert leaf.run_attempt == 2 assert leaf.variants is not None + assert leaf.variants[0].run_attempt == 2 assert leaf.variants[0].status is Status.success +def test_upsert_test_with_variants_always_returns_true() -> None: + # Mirrors test_upsert_build_with_variants_always_returns_true: the variant + # path merges per-cell and bypasses the leaf-level guard. + doc = StatusDocument() + assert doc.upsert_leaf( + "linux", + "gfx942", + "pytorch", + "test", + _leaf(variants=[_variant(matrix={"py": "3"})]), + ) + + +def test_upsert_test_merge_does_not_regress_completed_cell() -> None: + # This is the push-race scenario the fix targets: two concurrent + # `receive_therock_data.yml` runs fetch fresh job-list snapshots of the + # SAME shared entry run at different wall-clock times. Snapshot A sees + # py3.11 done but py3.12 still running; snapshot B (fetched slightly + # earlier, but whose git push lands second) sees py3.11 still running but + # py3.12 done. Regardless of push order, merging cell-by-cell must end up + # with BOTH cells advanced -- never one snapshot regressing the other's + # progress. + doc = StatusDocument() + run_id = 555 + doc.upsert_leaf( + "linux", + "gfx942", + "jax", + "test", + _leaf( + run_id=run_id, + run_attempt=1, + status=Status.in_progress, + completed_at=None, + variants=[ + _variant( + matrix={"py": "3.11"}, + run_id=run_id, + status=Status.success, + completed_at="2026-04-08T01:10:00Z", + ), + _variant( + matrix={"py": "3.12"}, + run_id=run_id, + status=Status.in_progress, + completed_at=None, + ), + ], + ), + ) + # A stale snapshot lands next: py3.11 looks in_progress again (fetched + # before it finished) but py3.12 has since completed. + doc.upsert_leaf( + "linux", + "gfx942", + "jax", + "test", + _leaf( + run_id=run_id, + run_attempt=1, + status=Status.in_progress, + completed_at=None, + variants=[ + _variant( + matrix={"py": "3.11"}, + run_id=run_id, + status=Status.in_progress, + completed_at=None, + ), + _variant( + matrix={"py": "3.12"}, + run_id=run_id, + status=Status.success, + completed_at="2026-04-08T01:20:00Z", + ), + ], + ), + ) + leaf = doc.pipelines.jax.test["linux"]["gfx942"] + assert leaf.variants is not None + by_key = {v.key(): v for v in leaf.variants} + assert by_key[(("py", "3.11"),)].status is Status.success + assert by_key[(("py", "3.12"),)].status is Status.success + # Both cells terminal -> the leaf-level rollup is terminal too. + assert leaf.status is Status.success + assert leaf.completed_at == "2026-04-08T01:20:00Z" + + def test_upsert_test_newer_run_id_supersedes_via_upsert() -> None: # A fresh re-dispatch mints a NEW workflow run (larger run_id). The guard # compares run_id FIRST, so the newer run wins regardless of attempt -- here diff --git a/scripts/receive_therock/therock_status_document.py b/scripts/receive_therock/therock_status_document.py index 1a0e5e8c97..68a9c572ff 100644 --- a/scripts/receive_therock/therock_status_document.py +++ b/scripts/receive_therock/therock_status_document.py @@ -368,6 +368,69 @@ def _merge_variant_leaf(existing: "RunLeaf | None", new: "RunLeaf") -> "RunLeaf" ) +def merge_matrix_test_leaf(existing: "RunLeaf | None", new: "RunLeaf") -> "RunLeaf": + """Merge fan-out test variants by matrix cell instead of replacing the + whole leaf (mirrors `_merge_variant_leaf`, but also carries forward + `run_id`/`run_attempt`/timestamps so later comparisons -- e.g. + `_refresh_same_run_fanout_tests` in therock_update_status_json.py, its + other caller -- can still identify the owning run). + + Every completion notification for a shared-entry-run matrix (pytorch/jax + py x ref fan-out sharing one `GITHUB_RUN_ID`) re-derives *all* cells from + a freshly re-fetched job-list snapshot, and those snapshots race each + other under heavy concurrency with no correlation between "wins the push" + and "is the freshest/most complete". Wholesale replacement + (`arch_map[arch] = new`) lets an earlier, less-complete snapshot silently + regress a later, more-complete one whenever it happens to win the race. + Merging cell-by-cell through `Variant.should_replace` makes every cell + advance monotonically regardless of push-race ordering. + """ + variants: list[Variant] = [] + positions: dict[tuple[tuple[str, str], ...], int] = {} + + for variant in existing.variants if existing and existing.variants else []: + positions[variant.key()] = len(variants) + variants.append(variant) + + for variant in new.variants or []: + key = variant.key() + pos = positions.get(key) + if pos is None: + positions[key] = len(variants) + variants.append(variant) + elif variants[pos].should_replace(variant): + variants[pos] = variant + + status = Variant.rollup_status(variants, new.status) + completed_at: str | None = None + if status.is_terminal: + ends = [v.completed_at for v in variants if v.completed_at] + completed_at = max(ends) if ends else new.completed_at + + starts = [v.started_at for v in variants if v.started_at] + started_at = starts and min(starts) + if not started_at: + started_at = (existing.started_at if existing else None) or new.started_at + + run_id = ( + new.run_id if new.run_id is not None else (existing.run_id if existing else None) + ) + run_attempt = ( + new.run_attempt + if new.run_attempt is not None + else (existing.run_attempt if existing else None) + ) + + return RunLeaf( + status=status, + run_id=run_id, + run_attempt=run_attempt, + started_at=started_at or None, + completed_at=completed_at, + variants=variants, + ) + + # --- Summary rollup ---------------------------------------------------------- # `therock_summary.rebuild_summary` derives these from the pipeline detail tree # on every update; they are the read-optimized projection consumers render. @@ -581,6 +644,9 @@ def upsert_leaf( ) arch_map = phase_map.setdefault(platform, {}) existing = arch_map.get(arch) + if leaf.variants: + arch_map[arch] = merge_matrix_test_leaf(existing, leaf) + return True if existing is not None and not existing.should_replace(leaf): return False arch_map[arch] = leaf diff --git a/scripts/receive_therock/therock_update_status_json.py b/scripts/receive_therock/therock_update_status_json.py index cc74af9105..fec6b655cb 100644 --- a/scripts/receive_therock/therock_update_status_json.py +++ b/scripts/receive_therock/therock_update_status_json.py @@ -56,6 +56,7 @@ Status, StatusDocument, Variant, + merge_matrix_test_leaf, rollup_statuses, ) from therock_summary import freeze_requested_architectures, rebuild_summary @@ -528,8 +529,13 @@ def _refresh_same_run_fanout_tests( started). Fold it into the rollup rather than only using it as an empty-variants fallback, so a terminal failure/cancellation at the run level cannot be masked by whatever the individual cells happened to report - -- mirroring what `_merge_matrix_build_leaf` does for the build leaf - itself. + -- mirroring what `_merge_variant_leaf` does for the build leaf itself. + + Merges cell-by-cell (via `merge_matrix_test_leaf`) rather than replacing + the leaf's `variants` wholesale: this snapshot's own variants are derived + from the build run's job names, so a blind overwrite could clobber + genuinely newer/more-complete per-cell results that already landed from + the test leaf's own dedicated completion events. """ cls = workflow_run.classification if ( @@ -548,16 +554,17 @@ def _refresh_same_run_fanout_tests( wrote = False for phase_map in (pipeline.test, pipeline.test_full): for arch_map in phase_map.values(): - for existing in arch_map.values(): + for arch, existing in list(arch_map.items()): if existing.run_id != leaf.run_id: continue if (existing.run_attempt or 0) != (leaf.run_attempt or 0): continue if not existing.should_replace(leaf): continue - existing.status = projected_status - existing.completed_at = leaf.completed_at - existing.variants = leaf.variants + merged = merge_matrix_test_leaf(existing, leaf) + merged.status = projected_status + merged.completed_at = leaf.completed_at + arch_map[arch] = merged wrote = True return wrote From edd4feb863de22a24a364b0266679826b6473ecd Mon Sep 17 00:00:00 2001 From: Laura Promberger Date: Mon, 17 Aug 2026 12:51:57 +0200 Subject: [PATCH 2/2] use should_replace() to calculate new run id and run attempt for merge_matrix_test_leaf. add test for it. small fix for code readability. --- .../tests/therock_status_document_test.py | 57 +++++++++++++++++++ .../therock_status_document.py | 14 ++--- 2 files changed, 62 insertions(+), 9 deletions(-) diff --git a/scripts/receive_therock/tests/therock_status_document_test.py b/scripts/receive_therock/tests/therock_status_document_test.py index 36abf609d3..3333d14f42 100644 --- a/scripts/receive_therock/tests/therock_status_document_test.py +++ b/scripts/receive_therock/tests/therock_status_document_test.py @@ -449,6 +449,63 @@ def test_merge_status_rolls_up_failure() -> None: assert merged.status is Status.failure +# --- merge_matrix_test_leaf: leaf-header identity --------------------------- + + +def test_matrix_merge_leaf_header_keeps_newer_run_over_stale_loser() -> None: + # A stale older-run snapshot loses the push race but still reaches the + # merge (the variant path bypasses RunLeaf.should_replace). Cells are held + # by Variant.should_replace; the leaf header must not regress with them. + existing = _leaf( + run_id=200, + run_attempt=1, + variants=[_variant(matrix={"py": "3.11"}, run_id=200, run_attempt=1)], + ) + new = _leaf( + run_id=100, + run_attempt=5, + variants=[_variant(matrix={"py": "3.11"}, run_id=100, run_attempt=5)], + ) + merged = merge_matrix_test_leaf(existing, new) + assert merged.run_id == 200 + assert merged.run_attempt == 1 + assert merged.variants is not None + assert merged.variants[0].run_id == 200 + + +def test_matrix_merge_leaf_header_advances_to_newer_run() -> None: + existing = _leaf( + run_id=100, + run_attempt=2, + variants=[_variant(matrix={"py": "3.11"}, run_id=100, run_attempt=2)], + ) + new = _leaf( + run_id=200, + run_attempt=1, + variants=[_variant(matrix={"py": "3.11"}, run_id=200, run_attempt=1)], + ) + merged = merge_matrix_test_leaf(existing, new) + # newer run_id wins outright -- its (lower) attempt comes with it. + assert merged.run_id == 200 + assert merged.run_attempt == 1 + + +def test_matrix_merge_leaf_header_takes_higher_attempt_within_run() -> None: + existing = _leaf( + run_id=100, + run_attempt=2, + variants=[_variant(matrix={"py": "3.11"}, run_id=100, run_attempt=2)], + ) + new = _leaf( + run_id=100, + run_attempt=1, + variants=[_variant(matrix={"py": "3.11"}, run_id=100, run_attempt=1)], + ) + merged = merge_matrix_test_leaf(existing, new) + assert merged.run_id == 100 + assert merged.run_attempt == 2 + + # --- upsert_leaf: build phase ----------------------------------------------- diff --git a/scripts/receive_therock/therock_status_document.py b/scripts/receive_therock/therock_status_document.py index 68a9c572ff..c53de9e069 100644 --- a/scripts/receive_therock/therock_status_document.py +++ b/scripts/receive_therock/therock_status_document.py @@ -408,18 +408,14 @@ def merge_matrix_test_leaf(existing: "RunLeaf | None", new: "RunLeaf") -> "RunLe completed_at = max(ends) if ends else new.completed_at starts = [v.started_at for v in variants if v.started_at] - started_at = starts and min(starts) + started_at = min(starts) if starts else None if not started_at: started_at = (existing.started_at if existing else None) or new.started_at - run_id = ( - new.run_id if new.run_id is not None else (existing.run_id if existing else None) - ) - run_attempt = ( - new.run_attempt - if new.run_attempt is not None - else (existing.run_attempt if existing else None) - ) + if existing is None or existing.should_replace(new): + run_id, run_attempt = new.run_id, new.run_attempt + else: + run_id, run_attempt = existing.run_id, existing.run_attempt return RunLeaf( status=status,