From 6ad54959bd8dc8ee0d28f4e596ad3797963c8a21 Mon Sep 17 00:00:00 2001 From: smngvlkz Date: Sat, 10 Oct 2026 10:30:49 +0200 Subject: [PATCH] fix(watch): don't drop full rebuilds when the lock is busy (#4274) --- graphify/watch.py | 93 ++++++++++++++++++++++++-------------- tests/test_watch.py | 107 +++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 164 insertions(+), 36 deletions(-) diff --git a/graphify/watch.py b/graphify/watch.py index 306d0694f..adac27357 100644 --- a/graphify/watch.py +++ b/graphify/watch.py @@ -21,6 +21,7 @@ from graphify.paths import ( logger = logging.getLogger(__name__) _PENDING_FILENAME = ".pending_changes" +_PENDING_FULL_FILENAME = ".pending_full" _PENDING_DRAIN_MAX_PASSES = 20 @@ -78,6 +79,26 @@ def _drain_pending(out_dir: Path) -> list[Path]: return out +def _queue_pending_full(out_dir: Path) -> None: + """Record that a full-corpus rebuild was requested but could not run. + + Full requests (watcher, post-checkout hook) have no path list to append to + ``.pending_changes``, so they leave a marker file instead. The lock-holder + consumes it and runs one more full pass before releasing (#4274). + """ + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / _PENDING_FULL_FILENAME).touch() + + +def _drain_pending_full(out_dir: Path) -> bool: + """Consume the full-rebuild marker. Returns True if one was queued.""" + try: + (out_dir / _PENDING_FULL_FILENAME).unlink() + except FileNotFoundError: + return False + return True + + # Build options that must survive into later rebuilds. The initial `extract` # scan honours `--exclude`, but `update`/`watch`/hook rebuilds re-run detect() # and would silently re-include excluded paths unless the patterns are persisted @@ -1593,29 +1614,32 @@ def _rebuild_code( out = watch_path / _GRAPHIFY_OUT if acquire_lock: - # #1059: incremental (changed_paths is not None) hooks must not drop - # their change set when another rebuild is already running. Queue - # before attempting the lock so a non-blocking failure still records - # the work; the lock-holder drains the queue and merges it in. Full- - # corpus rebuilds skip the queue entirely — they already cover every - # file, so there is nothing to merge. - if changed_paths is not None and not block_on_lock: - _queue_pending(out, list(changed_paths)) + # #1059: a hook must not drop its request when another rebuild is + # already running. Queue before attempting the lock so a non-blocking + # failure still records the work; the lock-holder drains the queue. + # Incremental requests append their paths to .pending_changes; full- + # corpus requests leave a .pending_full marker (#4274). + queued = False + if not block_on_lock: + if changed_paths is None: + _queue_pending_full(out) + queued = True + elif changed_paths: + _queue_pending(out, list(changed_paths)) + queued = True with _rebuild_lock(out, blocking=block_on_lock) as got: if not got: print("[graphify watch] Rebuild already in progress for " - f"{watch_path.resolve()} - changes queued.") + f"{watch_path.resolve()} - " + f"{'changes queued' if queued else 'nothing to queue'}.") return False # Lock acquired. Drain anything queued by earlier contenders - # (including, importantly, the paths we just queued ourselves) - # and merge with our own change set so a single rebuild covers - # everything outstanding. - if changed_paths is not None: - merged = _merge_changed_paths(changed_paths, _drain_pending(out)) - else: - # Full-corpus rebuild supersedes any queued incremental work. - _drain_pending(out) - merged = None + # (including, importantly, what we just queued ourselves). A + # queued full request upgrades this run to a full rebuild, which + # supersedes any incremental work queued before it starts. + full = _drain_pending_full(out) or changed_paths is None + pending = _drain_pending(out) + merged = None if full else _merge_changed_paths(changed_paths, pending) ok = _rebuild_code( watch_path, changed_paths=merged, @@ -1625,22 +1649,23 @@ def _rebuild_code( acquire_lock=False, ) # Late-arrival drain: another hook may have queued work while we - # were rebuilding. Loop up to _PENDING_DRAIN_MAX_PASSES times so a - # storm of commits eventually quiesces without livelocking. A full - # rebuild already saw everything, so skip this for changed_paths is None. - if merged is not None: - for _ in range(_PENDING_DRAIN_MAX_PASSES): - late = _drain_pending(out) - if not late: - break - ok = _rebuild_code( - watch_path, - changed_paths=late, - follow_symlinks=follow_symlinks, - force=force, - no_cluster=no_cluster, - acquire_lock=False, - ) and ok + # were rebuilding — even a full rebuild only saw files as they were + # when it walked the corpus (#4274). Loop up to + # _PENDING_DRAIN_MAX_PASSES times so a storm of commits eventually + # quiesces without livelocking. + for _ in range(_PENDING_DRAIN_MAX_PASSES): + late_full = _drain_pending_full(out) + late = _drain_pending(out) + if not late_full and not late: + break + ok = _rebuild_code( + watch_path, + changed_paths=None if late_full else late, + follow_symlinks=follow_symlinks, + force=force, + no_cluster=no_cluster, + acquire_lock=False, + ) and ok return ok watch_root = watch_path.resolve() diff --git a/tests/test_watch.py b/tests/test_watch.py index eea145376..56909c98b 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -2161,8 +2161,111 @@ def test_rebuild_code_full_corpus_skips_pending_queue(tmp_path, monkeypatch): # Full-corpus rebuild passes None to the inner call (does not merge in # the queued paths — a full rebuild already covers them). assert seen == [None] - # The queue still gets drained on entry so stale entries don't leak, - # but no late-arrival loop runs for the full-corpus path. + # The queue is drained on entry so stale entries don't leak; nothing + # arrived mid-run, so the late-arrival loop has nothing to do. + assert not (out / watch_mod._PENDING_FILENAME).exists() + + +def test_rebuild_code_full_request_queues_marker_on_contention(tmp_path, capsys): + """#4274: a full-corpus request that loses the lock must leave a + .pending_full marker instead of being dropped behind a 'queued' log.""" + from graphify.watch import _rebuild_code, _rebuild_lock, _PENDING_FULL_FILENAME + + out = tmp_path / "graphify-out" + out.mkdir() + + with _rebuild_lock(out, blocking=False) as outer_got: + assert outer_got is True + assert _rebuild_code(tmp_path) is False + assert "changes queued" in capsys.readouterr().out + assert (out / _PENDING_FULL_FILENAME).exists() + + +def test_rebuild_code_empty_change_set_does_not_claim_queued(tmp_path, capsys): + """#4274: 'changes queued' is printed only when something was queued.""" + from graphify.watch import _rebuild_code, _rebuild_lock, _PENDING_FILENAME + + out = tmp_path / "graphify-out" + out.mkdir() + + with _rebuild_lock(out, blocking=False) as outer_got: + assert outer_got is True + assert _rebuild_code(tmp_path, changed_paths=[]) is False + assert "changes queued" not in capsys.readouterr().out + assert not (out / _PENDING_FILENAME).exists() + + +def test_rebuild_code_holder_runs_queued_full_pass(tmp_path, monkeypatch): + """#4274: a full request queued while an incremental rebuild holds the + lock must make the holder run one more full pass before releasing.""" + from graphify import watch as watch_mod + from graphify.watch import _rebuild_code as orig_rebuild + + out = tmp_path / "graphify-out" + out.mkdir() + seen: list = [] + + def fake_inner(watch_path, **kwargs): + if kwargs.get("acquire_lock") is False: + paths = kwargs.get("changed_paths") + seen.append(None if paths is None else [p.as_posix() for p in paths]) + if len(seen) == 1: + watch_mod._queue_pending_full(out) + return True + + monkeypatch.setattr(watch_mod, "_rebuild_code", fake_inner) + + assert orig_rebuild(tmp_path, changed_paths=[Path("own.py")]) is True + assert seen == [["own.py"], None] + assert not (out / watch_mod._PENDING_FULL_FILENAME).exists() + + +def test_rebuild_code_full_pass_upgrades_from_queued_marker(tmp_path, monkeypatch): + """#4274: an incremental holder that finds a full marker already queued + runs a full rebuild instead of its incremental one.""" + from graphify import watch as watch_mod + from graphify.watch import _rebuild_code as orig_rebuild + + out = tmp_path / "graphify-out" + out.mkdir() + watch_mod._queue_pending_full(out) + seen: list = [] + + def fake_inner(watch_path, **kwargs): + if kwargs.get("acquire_lock") is False: + seen.append(kwargs.get("changed_paths")) + return True + + monkeypatch.setattr(watch_mod, "_rebuild_code", fake_inner) + + assert orig_rebuild(tmp_path, changed_paths=[Path("own.py")]) is True + assert seen == [None] + assert not (out / watch_mod._PENDING_FULL_FILENAME).exists() + assert not (out / watch_mod._PENDING_FILENAME).exists() + + +def test_rebuild_code_full_rebuild_drains_late_arrivals(tmp_path, monkeypatch): + """#4274: incremental work queued after a full rebuild walked the corpus + must be applied by that rebuild's holder before it releases the lock.""" + from graphify import watch as watch_mod + from graphify.watch import _rebuild_code as orig_rebuild + + out = tmp_path / "graphify-out" + out.mkdir() + seen: list = [] + + def fake_inner(watch_path, **kwargs): + if kwargs.get("acquire_lock") is False: + paths = kwargs.get("changed_paths") + seen.append(None if paths is None else [p.as_posix() for p in paths]) + if len(seen) == 1: + watch_mod._queue_pending(out, [Path("late.py")]) + return True + + monkeypatch.setattr(watch_mod, "_rebuild_code", fake_inner) + + assert orig_rebuild(tmp_path) is True + assert seen == [None, ["late.py"]] assert not (out / watch_mod._PENDING_FILENAME).exists()