Skip to content

fix(workspace): single-flight workspace updates #1834

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sloemo01 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(workspace): single-flight workspace updates #1834

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
205 changes: 131 additions & 74 deletions packages/core/src/repowise/core/update_lock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()),
Expand All @@ -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)
Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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"

Loading
Loading