state_lock

class StateLock(KeyLock):
    """Exclusive access to the state ``key``: read its version, or seal and repoint a new one."""

    def __init__(self, scratch: Scratch, key: str, timeout_seconds: float) -> None:
        """Acquire the lock on ``key``, waiting up to ``timeout_seconds``.

        Raises:
            RecipeError: If the lock is still held elsewhere when the timeout passes.
        """
        self._scratch = scratch
        super().__init__(scratch.state_locks_dir, key)
        self.acquire(timeout_seconds)

    def get(self) -> tuple[Path, ContentHash] | None:
        """The current version of the state as ``(object directory, hash)``, or ``None``."""
        with self._scratch.database.transaction() as db:
            hash = names.resolve(db, names.state_name(self.key))
        if hash is None or not self._scratch.object_dir(hash).is_dir():
            return None
        return self._scratch.object_dir(hash), hash

    def swap(self, workspace: Workspace, stem: str) -> tuple[Path, ContentHash]:
        """Seal ``workspace/<stem>`` as the state's new version and make it current.

        Raises:
            RecipeError: If this lock has been released.
        """
        if not self.held:
            raise RecipeError(f"state lock on {self.key!r} is no longer held")
        hash = seal_tree(self._scratch, workspace.path / stem)
        with self._scratch.database.transaction() as db:
            names.point(db, names.state_name(self.key), hash, kept=True)
        workspace.release()
        return self._scratch.object_dir(hash), hash