feat(fs): expose isLocked in stat() to surface path-lock state (#1940)

Adds an `isLocked` boolean to the dict returned by `VikingFS.stat()` so
callers (and the `/api/v1/fs/stat` endpoint, which transparently passes
the dict through) can tell whether a resource is currently held by a
path lock without having to attempt a write and observe `ResourceBusyError`.

The lookup reuses the same conflict-detection semantics as the acquire
flow: a path is reported as locked when it has a valid `.path.ovlock` or
when any ancestor directory holds a SUBTREE lock; stale locks are
ignored because the next acquirer would reclaim them anyway.

To make the check available to higher layers, a public
`PathLock.is_locked()` helper is introduced and surfaced through
`LockManager.is_path_locked()`; both are best-effort and degrade to
`False` when the LockManager is unavailable, keeping `stat()` resilient.

Co-authored-by: dingben.db@bytedance.com <dingben.db@bytedance.com@bytedance.com>

Author: fengluodb

openviking/storage/transaction/lock_manager.py

@@ -197,6 +197,20 @@ def get_handle(self, handle_id: str) -> Optional[LockHandle]:
             return None
         return current
 
+    def is_path_locked(self, path: str, ignore_stale: bool = True) -> bool:
+        """Check whether *path* is currently locked.
+
+        Semantics align with conflict detection in the acquire flow: the path
+        is considered locked if it (or any ancestor) holds a lock. By default,
+        stale locks are ignored because they will be reclaimed on the next
+        acquire attempt.
+        """
+        try:
+            return self._path_lock.is_locked(path, ignore_stale=ignore_stale)
+        except Exception as e:
+            logger.warning(f"is_path_locked failed for {path}: {e}")
+            return False
+
     async def refresh_lock(self, handle: LockHandle) -> None:
         current = self._reconcile_handle(handle)
         if current is None:

openviking/storage/transaction/path_lock.py

@@ -115,6 +115,42 @@ def is_lock_owned_by(self, lock_path: str, owner_id: str) -> bool:
         current_owner_id, _ = self._read_owner_and_type(lock_path)
         return current_owner_id == owner_id
 
+    def is_locked(self, path: str, ignore_stale: bool = True) -> bool:
+        """Check whether *path* is currently locked.
+
+        Detection rules (aligned with conflict checks in the acquire flow):
+        - The path itself has a valid .path.ovlock; or
+        - Any ancestor directory holds a SUBTREE lock.
+
+        Args:
+            path: Path to check (already converted to AGFS internal path).
+            ignore_stale: Whether to ignore expired (stale) locks. Defaults
+                to True to stay consistent with the acquire flow: stale
+                locks will be cleaned up by the next acquirer, so they are
+                not considered as held here.
+        """
+        # 1. Lock on the path itself
+        own_lock_path = self._get_lock_path(path)
+        token = self._read_token(own_lock_path)
+        if token is not None:
+            if not (ignore_stale and self.is_lock_stale(own_lock_path, self._lock_expire)):
+                return True
+
+        # 2. Ancestor SUBTREE locks
+        parent = self._get_parent_path(path)
+        while parent:
+            ancestor_lock = self._get_lock_path(parent)
+            ancestor_token = self._read_token(ancestor_lock)
+            if ancestor_token is not None:
+                _, _, lock_type = _parse_fencing_token(ancestor_token)
+                if lock_type == LOCK_TYPE_SUBTREE and not (
+                    ignore_stale and self.is_lock_stale(ancestor_lock, self._lock_expire)
+                ):
+                    return True
+            parent = self._get_parent_path(parent)
+
+        return False
+
     def collect_lost_owner_locks(self, owner: LockOwner) -> list[str]:
         lost_paths: list[str] = []
         for lock_path in list(owner.locks):

openviking/storage/viking_fs.py

@@ -945,11 +945,29 @@ async def stat(self, uri: str, ctx: Optional[RequestContext] = None) -> Dict[str
         """
         File/directory information.
 
-        example: {'name': 'resources', 'size': 128, 'mode': 2147484141, 'modTime': '2026-02-10T21:26:02.934376379+08:00', 'isDir': True, 'meta': {'Name': 'localfs', 'Type': 'local', 'Content': {'local_path': '...'}}}
+        example: {'name': 'resources', 'size': 128, 'mode': 2147484141, 'modTime': '2026-02-10T21:26:02.934376379+08:00', 'isDir': True, 'isLocked': False, 'meta': {'Name': 'localfs', 'Type': 'local', 'Content': {'local_path': '...'}}}
+
+        Extra field:
+            isLocked (bool): Whether the path is currently held by a path lock
+                (either the path itself or any ancestor directory). Returns
+                False when the LockManager is not initialized or the lookup
+                fails.
         """
         self._ensure_access(uri, ctx)
         path = self._uri_to_path(uri, ctx=ctx)
-        return self.agfs.stat(path)
+        result = self.agfs.stat(path)
+        if isinstance(result, dict):
+            result["isLocked"] = self._is_path_locked(path)
+        return result
+
+    def _is_path_locked(self, path: str) -> bool:
+        """Best-effort path-lock lookup; returns False when LockManager is absent."""
+        try:
+            from openviking.storage.transaction import get_lock_manager
+
+            return get_lock_manager().is_path_locked(path)
+        except Exception:
+            return False
 
     async def exists(self, uri: str, ctx: Optional[RequestContext] = None) -> bool:
         """Check if a URI exists.

tests/transaction/test_path_lock.py

@@ -73,6 +73,58 @@ def test_is_lock_stale_recent_token(self):
         assert lock.is_lock_stale("/test/.path.ovlock", expire_seconds=300.0) is False
 
 
+class TestPathLockIsLocked:
+    def _agfs_with_locks(self, locks: dict[str, bytes]) -> MagicMock:
+        """Build an AGFS mock that only returns content for paths in *locks*."""
+        agfs = MagicMock()
+
+        def _read(p):
+            if p in locks:
+                return locks[p]
+            raise Exception("not found")
+
+        agfs.read.side_effect = _read
+        return agfs
+
+    def test_is_locked_no_lock(self):
+        agfs = self._agfs_with_locks({})
+        lock = PathLock(agfs)
+        assert lock.is_locked("/local/u/foo") is False
+
+    def test_is_locked_self_point_lock(self):
+        token = _make_fencing_token("tx-1", LOCK_TYPE_POINT)
+        agfs = self._agfs_with_locks({"/local/u/foo/.path.ovlock": token.encode("utf-8")})
+        lock = PathLock(agfs)
+        assert lock.is_locked("/local/u/foo") is True
+
+    def test_is_locked_ancestor_subtree_lock(self):
+        token = _make_fencing_token("tx-1", LOCK_TYPE_SUBTREE)
+        agfs = self._agfs_with_locks({"/local/u/.path.ovlock": token.encode("utf-8")})
+        lock = PathLock(agfs)
+        assert lock.is_locked("/local/u/foo/bar") is True
+
+    def test_is_locked_ancestor_point_lock_does_not_propagate(self):
+        """An ancestor POINT lock must not affect descendants -- POINT locks only the path itself."""
+        token = _make_fencing_token("tx-1", LOCK_TYPE_POINT)
+        agfs = self._agfs_with_locks({"/local/u/.path.ovlock": token.encode("utf-8")})
+        lock = PathLock(agfs)
+        assert lock.is_locked("/local/u/foo/bar") is False
+
+    def test_is_locked_ignores_stale_by_default(self):
+        old_ts = time.time_ns() - int(600 * 1e9)  # 600s ago
+        stale = f"tx-dead:{old_ts}:{LOCK_TYPE_POINT}".encode("utf-8")
+        agfs = self._agfs_with_locks({"/local/u/foo/.path.ovlock": stale})
+        lock = PathLock(agfs, lock_expire=300.0)
+        assert lock.is_locked("/local/u/foo") is False
+
+    def test_is_locked_can_include_stale(self):
+        old_ts = time.time_ns() - int(600 * 1e9)
+        stale = f"tx-dead:{old_ts}:{LOCK_TYPE_POINT}".encode("utf-8")
+        agfs = self._agfs_with_locks({"/local/u/foo/.path.ovlock": stale})
+        lock = PathLock(agfs, lock_expire=300.0)
+        assert lock.is_locked("/local/u/foo", ignore_stale=False) is True
+
+
 class TestPathLockOwnership:
     async def test_refresh_reports_refreshed_lost_and_failed_paths(self):
         owned_path = "/locks/owned/.path.ovlock"