diff --git a/packages/core/src/repowise/core/update_lock.py b/packages/core/src/repowise/core/update_lock.py index a2d56f42c..b71eb00ab 100644 --- a/packages/core/src/repowise/core/update_lock.py +++ b/packages/core/src/repowise/core/update_lock.py @@ -42,29 +42,31 @@ def update_lock_path(repo_path: Path) -> Path: return Path(repo_path) / ".repowise" / UPDATE_LOCK_FILENAME -def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[str, Any] | None: - """Atomically acquire the update lock. ``None`` means acquired. +def workspace_update_lock_path(workspace_root: Path) -> Path: + """Path of the workspace-level single-flight update lock. - Returns the live owner's payload when another update already holds the - lock, so the caller can report who it lost to and bail. The payload is - written to a private temp file and hard-linked into place, so the lock - only ever becomes visible with its full content — an exclusive create - followed by a write leaves a window where a contender reads the still - empty file, mistakes it for a corrupt lock, and deletes the winner's - live lock (two "winners"). A stale lock (dead or recycled PID, or past - the wall-clock ceiling) is cleared and the create retried. + Unlike :func:`update_lock_path` (per-repo), this guards the whole + workspace so a rebase's N post-commit hooks coalesce into one full + ``update_workspace`` pass instead of N redundant ones. Lives under the + workspace data dir (``.repowise-workspace/``) so it is shared by every + member repo's update. + """ + return Path(workspace_root) / ".repowise-workspace" / UPDATE_LOCK_FILENAME - The payload contains the PID and target commit so the augment hook can - decide whether a stale-wiki warning is redundant, plus the writing - process's creation-time token so ``read_update_lock`` can tell a live - lock owner apart from an unrelated process that recycled the PID. - Best-effort: unexpected ``OSError`` (read-only fs, permissions) counts - as acquired — the lock is advisory and must never block an update. - Callers must still call ``release_update_lock`` in a finally block. + +def _try_acquire_lock_at( + lock_path: Path, + target_commit: str | None, +) -> dict[str, Any] | None: + """Core exclusive-create acquire against an explicit lock path. + + Shared by the per-repo (:func:`try_acquire_update_lock`) and workspace + (:func:`update_workspace_lock`) single-flight guards — one + implementation for both, so the workspace guard inherits the same + crash/liveness/coalescing semantics as the per-repo one. """ from repowise.core.procutils import process_create_token - lock_path = update_lock_path(repo_path) payload = { "pid": os.getpid(), "pid_create_token": process_create_token(os.getpid()), @@ -75,6 +77,10 @@ def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[ tmp_path = lock_path.with_name( f"{UPDATE_LOCK_FILENAME}.{os.getpid()}.{threading.get_ident()}.tmp" ) + + def _read_existing() -> dict[str, Any] | None: + return _read_lock_at(lock_path) + for _ in range(2): try: lock_path.parent.mkdir(parents=True, exist_ok=True) @@ -84,7 +90,7 @@ def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[ # half-written file. os.link(tmp_path, lock_path) except FileExistsError: - existing = read_update_lock(repo_path) + existing = _read_existing() if existing is not None: return existing # Stale lock: clear it and retry the exclusive create. @@ -98,17 +104,17 @@ def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[ tmp_path.unlink(missing_ok=True) return None # Both create attempts lost a race against a stale lock that was then - # unlinked. Rather than falling through to a bare ``read_update_lock`` - # — which can return ``None`` ("acquired") with *no lock file on disk*, - # letting a caller proceed without holding the lock — make one final - # exclusive create. Winner: return ``None`` (owned). Loser: report the - # fresh winner. Still-unreadable degrades to acquired, as everywhere. + # unlinked. Rather than falling through to a bare read — which can return + # ``None`` ("acquired") with *no lock file on disk*, letting a caller + # proceed without holding the lock — make one final exclusive create. + # Winner: return ``None`` (owned). Loser: report the fresh winner. + # Still-unreadable degrades to acquired, as everywhere. try: lock_path.parent.mkdir(parents=True, exist_ok=True) tmp_path.write_text(data, encoding="utf-8") os.link(tmp_path, lock_path) except FileExistsError: - return read_update_lock(repo_path) + return _read_existing() except OSError: return None finally: @@ -117,68 +123,64 @@ def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[ return None -def release_update_lock(repo_path: Path) -> None: - """Remove the update lock file. Safe to call if it doesn't exist.""" - with contextlib.suppress(OSError): - update_lock_path(repo_path).unlink(missing_ok=True) +def try_acquire_update_lock(repo_path: Path, target_commit: str | None) -> dict[str, Any] | None: + """Atomically acquire the update lock. ``None`` means acquired. + Returns the live owner's payload when another update already holds the + lock, so the caller can report who it lost to and bail. The payload is + written to a private temp file and hard-linked into place, so the lock + only ever becomes visible with its full content — an exclusive create + followed by a write leaves a window where a contender reads the still + empty file, mistakes it for a corrupt lock, and deletes the winner's + live lock (two "winners"). A stale lock (dead or recycled PID, or past + the wall-clock ceiling) is cleared and the create retried. -def lock_age_seconds(payload: dict[str, Any] | None) -> float | None: - """Wall-clock age of a lock payload, or ``None`` when it cannot be told. + The payload contains the PID and target commit so the augment hook can + decide whether a stale-wiki warning is redundant, plus the writing + process's creation-time token so ``read_update_lock`` can tell a live + lock owner apart from an unrelated process that recycled the PID. + Best-effort: unexpected ``OSError`` (read-only fs, permissions) counts + as acquired — the lock is advisory and must never block an update. + Callers must still call ``release_update_lock`` in a finally block. + """ + return _try_acquire_lock_at(update_lock_path(repo_path), target_commit) - One implementation because every reporting site needs the same number and - each one deriving it separately is how the deferral message ended up - quoting no age at all. + +def update_workspace_lock(workspace_root: Path) -> dict[str, Any] | None: + """Acquire the workspace-level single-flight guard. ``None`` means acquired. + + See :func:`try_acquire_update_lock` for the semantics; this is the same + guard held against ``workspace_update_lock_path`` so two concurrent + ``update_workspace`` runs coalesce instead of both re-indexing every + member. """ - if not payload: - return None - started = payload.get("started_at") - if not isinstance(started, (int, float)): - return None - return max(0.0, time.time() - started) + return _try_acquire_lock_at(workspace_update_lock_path(workspace_root), None) -def format_lock_age(age: float | None) -> str: - """Human phrasing for how long a lock has been held. +def _release_lock_at(lock_path: Path) -> None: + with contextlib.suppress(OSError): + lock_path.unlink(missing_ok=True) - Takes the seconds rather than the payload so a caller that already carries - the age (a deferred repo result) does not have to rebuild a payload to ask. - Coarsens with age on purpose: a lock held for seconds is normal and the - seconds are the interesting part, while one held for hours is the whole - point of the message and "32700s" buries it. - """ - if age is None: - return "for an unknown time" - if age < 90: - return f"for {int(age)}s" - if age < 90 * 60: - return f"for {int(age / 60)}m" - return f"for {age / 3600:.1f}h" +def release_update_lock(repo_path: Path) -> None: + """Remove the per-repo update lock file. Safe to call if it doesn't exist.""" + _release_lock_at(update_lock_path(repo_path)) -def read_update_lock(repo_path: Path) -> dict[str, Any] | None: - """Return the lock payload if present and not stale, else ``None``. +def release_workspace_lock(workspace_root: Path) -> None: + """Remove the workspace-level update lock. Safe to call if it doesn't exist.""" + _release_lock_at(workspace_update_lock_path(workspace_root)) - A lock is stale when its owning PID is positively dead or has been - recycled by an unrelated process. That probe is what stops a crashed or - killed update (SIGKILL, power loss — paths atexit cannot cover) from - blocking every later update. - An owner we can positively see running is honoured no matter how old the - lock is. The wall clock applies only when liveness cannot be established: - a payload with no usable PID (written by an older version) or a probe that - returned "unknown". Age on its own is not evidence that an update has - stopped: a full update on a large repo can outrun any ceiling worth - setting, and clearing the lock underneath it would put two updates on one - index, both writing the same state and the same page rows. A live owner - that has held the lock unreasonably long is surfaced to the user by the - callers instead (see :func:`lock_is_suspect`), which is the reporting half - of the same problem and cannot corrupt anything. +def _read_lock_at(lock_path: Path) -> dict[str, Any] | None: + """Read a lock payload from an explicit path, applying liveness/staleness. + + Mirrors :func:`read_update_lock` against an arbitrary lock path so the + workspace guard gets the same crash recovery: a dead or recycled owner's + lock is treated as absent, a live owner's is honored regardless of age. """ from repowise.core.procutils import pid_alive, process_create_token - lock_path = update_lock_path(repo_path) if not lock_path.exists(): return None try: @@ -197,8 +199,6 @@ def read_update_lock(repo_path: Path) -> dict[str, Any] | None: return None if alive is True: stored_token = payload.get("pid_create_token") - # Legacy locks (pre-token) skip the identity check: liveness is - # all we have, and it is still better evidence than the clock. if isinstance(stored_token, str) and stored_token: current_token = process_create_token(pid) if current_token is not None and current_token != stored_token: @@ -209,3 +209,60 @@ def read_update_lock(repo_path: Path) -> dict[str, Any] | None: if age > UPDATE_LOCK_STALE_AFTER_SECONDS: return None return payload + + +def read_update_lock(repo_path: Path) -> dict[str, Any] | None: + """Return the lock payload if present and not stale, else ``None``. + + A lock is stale when its owning PID is positively dead or has been + recycled by an unrelated process. That probe is what stops a crashed or + killed update (SIGKILL, power loss — paths atexit cannot cover) from + blocking every later update. + + An owner we can positively see running is honoured no matter how old the + lock is. The wall clock applies only when liveness cannot be established: + a payload with no usable PID (written by an older version) or a probe that + returned "unknown". Age on its own is not evidence that an update has + stopped: a full update on a large repo can outrun any ceiling worth + setting, and clearing the lock underneath it would put two updates on one + index, both writing the same state and the same page rows. A live owner + that has held the lock unreasonably long is surfaced to the user by the + callers instead (see :func:`lock_is_suspect`), which is the reporting half + of the same problem and cannot corrupt anything. + """ + return _read_lock_at(update_lock_path(repo_path)) + + +def lock_age_seconds(payload: dict[str, Any] | None) -> float | None: + """Wall-clock age of a lock payload, or ``None`` when it cannot be told. + + One implementation because every reporting site needs the same number and + each one deriving it separately is how the deferral message ended up + quoting no age at all. + """ + if not payload: + return None + started = payload.get("started_at") + if not isinstance(started, (int, float)): + return None + return max(0.0, time.time() - started) + + +def format_lock_age(age: float | None) -> str: + """Human phrasing for how long a lock has been held. + + Takes the seconds rather than the payload so a caller that already carries + the age (a deferred repo result) does not have to rebuild a payload to ask. + + Coarsens with age on purpose: a lock held for seconds is normal and the + seconds are the interesting part, while one held for hours is the whole + point of the message and "32700s" buries it. + """ + if age is None: + return "for an unknown time" + if age < 90: + return f"for {int(age)}s" + if age < 90 * 60: + return f"for {int(age / 60)}m" + return f"for {age / 3600:.1f}h" + diff --git a/packages/core/src/repowise/core/workspace/update.py b/packages/core/src/repowise/core/workspace/update.py index 2320946ed..6d115b403 100644 --- a/packages/core/src/repowise/core/workspace/update.py +++ b/packages/core/src/repowise/core/workspace/update.py @@ -33,9 +33,15 @@ from repowise.core.update_lock import ( release_update_lock as _release_lock, ) +from repowise.core.update_lock import ( + release_workspace_lock as _release_workspace_lock, +) from repowise.core.update_lock import ( try_acquire_update_lock as _try_acquire_lock, ) +from repowise.core.update_lock import ( + update_workspace_lock as _try_acquire_workspace_lock, +) from ..docs_mode import docs_mode_state_fields from ..ingestion.change_detector import has_working_tree_changes @@ -758,154 +764,198 @@ async def update_workspace( if dry_run or not stale_repos: return results - # Step 2: Update stale repos (parallel with concurrency limit) - semaphore = asyncio.Semaphore(_MAX_CONCURRENT_UPDATES) - - async def _update_one( - alias: str, path: Path, new_head: str, first_time: bool - ) -> RepoUpdateResult: - async with semaphore: - if on_repo_start: - on_repo_start(alias) - - # Ensure the .repowise/ dir exists before the pipeline runs so - # first-time indexing has a place to put wiki.db and state.json. - (path / ".repowise").mkdir(parents=True, exist_ok=True) - - # Per-repo single-flight lock. The post-commit hook fires a - # new ``repowise update`` for every commit; without this guard, - # rapid-fire commits race on save_state, each pass starts from - # the same stale base, and the wiki never converges to HEAD. - # Check + acquire are one atomic exclusive create. - existing = _try_acquire_lock(path, new_head) - if existing is not None: - age = _lock_age_seconds(existing) - target_short = (existing.get("target_commit") or "")[:8] - _log.info( - "workspace_update: skipping %s — update already in flight " - "(pid=%s target=%s elapsed=%ss)", - alias, - existing.get("pid"), - target_short, - int(age) if age is not None else "?", - ) - # Record pending so the running update can roll forward. + # Workspace-level single-flight guard. The per-repo lock below only + # stops two updates racing on the *same* repo's index; it does not stop + # N post-commit-hook ``repowise update`` invocations (one per rebase + # commit) from each running a full workspace pass over every stale + # member. Coalesce those triggers at the workspace level: the first pass + # holds this lock, and every concurrent one defers by recording a pending + # marker per stale repo so the running pass rolls forward to the latest + # HEAD instead of a second full index being spawned. + workspace_owner = _try_acquire_workspace_lock(workspace_root) + if workspace_owner is not None: + age = _lock_age_seconds(workspace_owner) + _log.info( + "workspace_update: deferring — a workspace update is already " + "in flight (pid=%s elapsed=%ss); queuing %d stale repo(s) for it", + workspace_owner.get("pid"), + int(age) if age is not None else "?", + len(stale_repos), + ) + # Record pending so the running workspace update rolls forward to the + # latest HEAD of each member. Mirrors the per-repo deferral in + # ``_update_one``. + for _alias, path, new_head, _first_time in stale_repos: + if new_head: with suppress(OSError): - (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8") - return RepoUpdateResult( - alias=alias, - updated=False, - skipped_reason="in_flight", - lock_age_seconds=age, - ) + (path / ".repowise").mkdir(parents=True, exist_ok=True) + (path / ".repowise" / ".update.pending").write_text( + new_head, encoding="utf-8" + ) + return [ + RepoUpdateResult( + alias=alias, + updated=False, + skipped_reason="in_flight", + lock_age_seconds=age, + ) + for alias, _path, _head, _first_time in stale_repos + ] - try: - result = await update_single_repo_index( - path, - commit_depth=commit_depth, - exclude_patterns=exclude_patterns, - include_working_tree=include_working_tree, - ) - finally: - _release_lock(path) - result.alias = alias - result.first_time_indexed = first_time and result.updated - - # Update state.json with new commit - if result.updated and new_head: - import json as _json - - state_path = path / ".repowise" / "state.json" - state: dict[str, Any] = {} - if state_path.is_file(): - with suppress(Exception): - state = _json.loads(state_path.read_text(encoding="utf-8")) - - if "last_docs_commit" not in state and "last_sync_commit" in state: - state["last_docs_commit"] = state["last_sync_commit"] - - state["last_sync_commit"] = new_head - if result.kg_state: - state["knowledge_graph"] = result.kg_state - if result.working_tree_paths is not None: - state["working_tree_paths"] = result.working_tree_paths - # Stamp the config fingerprint so the drift check in - # update_single_repo_index stays calibrated (and legacy repos - # without one stop re-triggering the full re-index). - with suppress(Exception): - from ..repo_config import config_fingerprint - - state["config_fingerprint"] = config_fingerprint(path) - # Mark first-time so downstream tooling (status, doctor) can - # distinguish a never-indexed repo from one that's been - # updated at least once. - if first_time and "docs_mode" not in state and "docs_enabled" not in state: - # This path indexes only; nothing renders pages here, not - # even from templates. - state.update(docs_mode_state_fields("none")) - state["docs_skip_reason"] = ( - "first-time index via update; run " - "`repowise update --repo " + alias + " --docs` to generate docs" + # Step 2: Update stale repos (parallel with concurrency limit). + # The workspace lock is held for the whole pass and released once every + # member has been attempted (including on failure) so the next hook-triggered + # invocation can take over. + try: + semaphore = asyncio.Semaphore(_MAX_CONCURRENT_UPDATES) + + async def _update_one( + alias: str, path: Path, new_head: str, first_time: bool + ) -> RepoUpdateResult: + async with semaphore: + if on_repo_start: + on_repo_start(alias) + + # Ensure the .repowise/ dir exists before the pipeline runs so + # first-time indexing has a place to put wiki.db and state.json. + (path / ".repowise").mkdir(parents=True, exist_ok=True) + + # Per-repo single-flight lock. The post-commit hook fires a + # new ``repowise update`` for every commit; without this guard, + # rapid-fire commits race on save_state, each pass starts from + # the same stale base, and the wiki never converges to HEAD. + # Check + acquire are one atomic exclusive create. + existing = _try_acquire_lock(path, new_head) + if existing is not None: + age = _lock_age_seconds(existing) + target_short = (existing.get("target_commit") or "")[:8] + _log.info( + "workspace_update: skipping %s — update already in flight " + "(pid=%s target=%s elapsed=%ss)", + alias, + existing.get("pid"), + target_short, + int(age) if age is not None else "?", + ) + # Record pending so the running update can roll forward. + with suppress(OSError): + (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8") + return RepoUpdateResult( + alias=alias, + updated=False, + skipped_reason="in_flight", + lock_age_seconds=age, ) - from ..fsutils import atomic_write_text - - state_path.parent.mkdir(parents=True, exist_ok=True) - atomic_write_text(state_path, _json.dumps(state, indent=2)) - # Keep the DB freshness stamp in lockstep with last_sync_commit. - # The no-relevant-changes incremental path returns updated=True - # without re-running DB persistence, so the row would otherwise - # lag HEAD; a no-op when persistence already stamped it. - await reconcile_repo_head_commit(path, new_head) - # Drop any stale pending marker now that we've advanced to - # new_head (a bailed sibling update may have written one). - clear_stale_update_pending(path, new_head) - - # Update workspace config entry - if result.updated: - entry = ws_config.get_repo(alias) - if entry is not None: - entry.indexed_at = datetime.now(UTC).isoformat() - entry.last_commit_at_index = new_head - - if on_repo_done: - on_repo_done(result) - - return result - - update_results = await asyncio.gather( - *[ - _update_one(alias, path, head, first_time) - for alias, path, head, first_time in stale_repos - ], - return_exceptions=True, - ) - changed_aliases: list[str] = [] - for r in update_results: - if isinstance(r, Exception): - results.append( - RepoUpdateResult( - alias="unknown", - updated=False, - error=str(r), + try: + result = await update_single_repo_index( + path, + commit_depth=commit_depth, + exclude_patterns=exclude_patterns, + include_working_tree=include_working_tree, + ) + finally: + _release_lock(path) + result.alias = alias + result.first_time_indexed = first_time and result.updated + + # Update state.json with new commit + if result.updated and new_head: + import json as _json + + state_path = path / ".repowise" / "state.json" + state: dict[str, Any] = {} + if state_path.is_file(): + with suppress(Exception): + state = _json.loads(state_path.read_text(encoding="utf-8")) + + if "last_docs_commit" not in state and "last_sync_commit" in state: + state["last_docs_commit"] = state["last_sync_commit"] + + state["last_sync_commit"] = new_head + if result.kg_state: + state["knowledge_graph"] = result.kg_state + if result.working_tree_paths is not None: + state["working_tree_paths"] = result.working_tree_paths + # Stamp the config fingerprint so the drift check in + # update_single_repo_index stays calibrated (and legacy repos + # without one stop re-triggering the full re-index). + with suppress(Exception): + from ..repo_config import config_fingerprint + + state["config_fingerprint"] = config_fingerprint(path) + # Mark first-time so downstream tooling (status, doctor) can + # distinguish a never-indexed repo from one that's been + # updated at least once. + if first_time and "docs_mode" not in state and "docs_enabled" not in state: + # This path indexes only; nothing renders pages here, not + # even from templates. + state.update(docs_mode_state_fields("none")) + state["docs_skip_reason"] = ( + "first-time index via update; run " + "`repowise update --repo " + alias + " --docs` to generate docs" + ) + from ..fsutils import atomic_write_text + + state_path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_text(state_path, _json.dumps(state, indent=2)) + # Keep the DB freshness stamp in lockstep with last_sync_commit. + # The no-relevant-changes incremental path returns updated=True + # without re-running DB persistence, so the row would otherwise + # lag HEAD; a no-op when persistence already stamped it. + await reconcile_repo_head_commit(path, new_head) + # Drop any stale pending marker now that we've advanced to + # new_head (a bailed sibling update may have written one). + clear_stale_update_pending(path, new_head) + + # Update workspace config entry + if result.updated: + entry = ws_config.get_repo(alias) + if entry is not None: + entry.indexed_at = datetime.now(UTC).isoformat() + entry.last_commit_at_index = new_head + + if on_repo_done: + on_repo_done(result) + + return result + + update_results = await asyncio.gather( + *[ + _update_one(alias, path, head, first_time) + for alias, path, head, first_time in stale_repos + ], + return_exceptions=True, + ) + + changed_aliases: list[str] = [] + for r in update_results: + if isinstance(r, Exception): + results.append( + RepoUpdateResult( + alias="unknown", + updated=False, + error=str(r), + ) ) - ) - else: - results.append(r) - if r.updated: - changed_aliases.append(r.alias) - - # Step 3: Save workspace config with updated timestamps - if changed_aliases: - ws_config.save(workspace_root) - - # Step 4: Run cross-repo hooks (Phase 3/4 placeholder). ``run_hooks`` lets - # the CLI defer these so they run once over the union of index-only and - # docs repos, rather than on this partial set. - if changed_aliases and run_hooks: - await run_cross_repo_hooks(ws_config, workspace_root, changed_aliases) - - return results + else: + results.append(r) + if r.updated: + changed_aliases.append(r.alias) + + # Step 3: Save workspace config with updated timestamps + if changed_aliases: + ws_config.save(workspace_root) + + # Step 4: Run cross-repo hooks (Phase 3/4 placeholder). ``run_hooks`` lets + # the CLI defer these so they run once over the union of index-only and + # docs repos, rather than on this partial set. + if changed_aliases and run_hooks: + await run_cross_repo_hooks(ws_config, workspace_root, changed_aliases) + + return results + finally: + _release_workspace_lock(workspace_root) # --------------------------------------------------------------------------- diff --git a/tests/unit/cli/test_update_lock.py b/tests/unit/cli/test_update_lock.py index 814f5dc2d..a47d7c7ea 100644 --- a/tests/unit/cli/test_update_lock.py +++ b/tests/unit/cli/test_update_lock.py @@ -313,3 +313,72 @@ def test_workspace_uses_shared_core_lock(tmp_path: Path) -> None: assert payload["pid"] == os.getpid() finally: ws_update._release_lock(tmp_path) + + +# --------------------------------------------------------------------------- +# Workspace-level lock — the #1831 single-flight guard +# --------------------------------------------------------------------------- + + +def test_workspace_lock_lives_under_workspace_data_dir(tmp_path: Path) -> None: + """The workspace guard must NOT reuse the per-repo lock path — it lives in + ``.repowise-workspace/.update.lock`` so a workspace's members share one.""" + from repowise.core.update_lock import ( + update_workspace_lock, + workspace_update_lock_path, + ) + + path = workspace_update_lock_path(tmp_path) + assert path == tmp_path / ".repowise-workspace" / ".update.lock" + + assert update_workspace_lock(tmp_path) is None + try: + assert (tmp_path / ".repowise-workspace" / ".update.lock").exists() + # The per-repo lock is a different file — a held workspace lock does + # not block a single-repo update. + assert not (tmp_path / ".repowise" / ".update.lock").exists() + finally: + from repowise.core.update_lock import release_workspace_lock + + release_workspace_lock(tmp_path) + assert not (tmp_path / ".repowise-workspace" / ".update.lock").exists() + + +def test_workspace_lock_is_single_flight(tmp_path: Path) -> None: + """A second workspace update defers to the first instead of running a + redundant full pass.""" + from repowise.core.update_lock import ( + release_workspace_lock, + update_workspace_lock, + ) + + assert update_workspace_lock(tmp_path) is None + try: + owner = update_workspace_lock(tmp_path) + assert owner is not None + assert owner["pid"] == os.getpid() + finally: + release_workspace_lock(tmp_path) + + # After release a fresh acquire wins again. + assert update_workspace_lock(tmp_path) is None + release_workspace_lock(tmp_path) + + +def test_stale_workspace_lock_is_cleared(tmp_path: Path) -> None: + """A dead owner's workspace lock must not wedge the next update.""" + from repowise.core.update_lock import update_workspace_lock + from repowise.core.workspace.update import _release_workspace_lock + + ws_dir = tmp_path / ".repowise-workspace" + ws_dir.mkdir(parents=True, exist_ok=True) + (ws_dir / ".update.lock").write_text( + json.dumps({"pid": _dead_pid(), "target_commit": None, "started_at": time.time()}), + encoding="utf-8", + ) + + assert update_workspace_lock(tmp_path) is None + try: + assert (ws_dir / ".update.lock").exists() + finally: + _release_workspace_lock(tmp_path) diff --git a/tests/unit/workspace/test_update.py b/tests/unit/workspace/test_update.py index d90993c4b..c75820a4e 100644 --- a/tests/unit/workspace/test_update.py +++ b/tests/unit/workspace/test_update.py @@ -440,6 +440,91 @@ async def _run(): assert len(updated) == 0 +# --------------------------------------------------------------------------- +# Workspace-level single-flight (issue #1831) +# --------------------------------------------------------------------------- + + +class TestWorkspaceSingleFlight: + def test_concurrent_run_defers_to_owner(self, tmp_path: Path) -> None: + """A workspace update that finds another workspace update already in + flight must defer (skipped_reason=\"in_flight\") instead of running a + redundant full pass over every member.""" + repo = _make_git_repo(tmp_path, "backend") + old_head = get_head_commit(repo) + _write_state(repo, old_head) + _add_commit(repo, "new.txt") + + ws_config = WorkspaceConfig( + repos=[RepoEntry(path="backend", alias="backend", last_commit_at_index=old_head)], + default_repo="backend", + ) + ws_config.save(tmp_path) + + from repowise.core.update_lock import update_workspace_lock + + # Simulate another workspace update already in flight (same process). + assert update_workspace_lock(tmp_path) is None + try: + mock_result = RepoUpdateResult(alias="backend", updated=True, file_count=1, symbol_count=1) + + async def _run(): + with patch( + "repowise.core.workspace.update.update_single_repo_index", + new_callable=AsyncMock, + return_value=mock_result, + ): + return await update_workspace(tmp_path, ws_config) + + import asyncio + results = asyncio.run(_run()) + # No repo should have been updated — the whole pass deferred. + assert len(results) == 1 + assert results[0].updated is False + assert results[0].skipped_reason == "in_flight" + # The running owner should pick up the deferred head via the + # pending marker written by the defereer. + pending = (tmp_path / "backend" / ".repowise" / ".update.pending") + assert pending.exists() + assert pending.read_text(encoding="utf-8") == get_head_commit(tmp_path / "backend") + finally: + from repowise.core.update_lock import release_workspace_lock + + release_workspace_lock(tmp_path) + + def test_lock_released_after_pass(self, tmp_path: Path) -> None: + """update_workspace must release the workspace lock after a successful + pass so the next hook-triggered invocation can take over.""" + repo = _make_git_repo(tmp_path, "backend") + old_head = get_head_commit(repo) + _write_state(repo, old_head) + _add_commit(repo, "new.txt") + + ws_config = WorkspaceConfig( + repos=[RepoEntry(path="backend", alias="backend", last_commit_at_index=old_head)], + default_repo="backend", + ) + ws_config.save(tmp_path) + + mock_result = RepoUpdateResult(alias="backend", updated=True, file_count=1, symbol_count=1) + + from repowise.core.update_lock import workspace_update_lock_path + + async def _run(): + with patch( + "repowise.core.workspace.update.update_single_repo_index", + new_callable=AsyncMock, + return_value=mock_result, + ): + return await update_workspace(tmp_path, ws_config) + + import asyncio + asyncio.run(_run()) + + # The workspace lock was created during the pass and released at the end. + assert not workspace_update_lock_path(tmp_path).exists() + + # --------------------------------------------------------------------------- # Cross-repo hooks placeholder # ---------------------------------------------------------------------------