raise BrokenResourceError if lock owner has exited.

src/trio/_sync.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import inspect
 import math
 from typing import TYPE_CHECKING, Protocol
 
@@ -571,11 +572,19 @@ def acquire_nowait(self) -> None:
         """
 
         task = trio.lowlevel.current_task()
+
         if self._owner is task:
             raise RuntimeError("attempt to re-acquire an already held Lock")
         elif self._owner is None and not self._lot:
             # No-one owns it
             self._owner = task
+        elif (
+            self._owner is not None
+            and inspect.getcoroutinestate(self._owner.coro) is inspect.CORO_CLOSED
+        ):
+            raise trio.BrokenResourceError(
+                "Attempted acquire on Lock which will never be released, owner has exited.",
+            )
         else:
             raise trio.WouldBlock
 

src/trio/_tests/test_sync.py

@@ -6,6 +6,7 @@
 import pytest
 
 from .. import _core
+from .._core._tests.tutil import slow
 from .._sync import *
 from .._timeouts import sleep_forever
 from ..testing import assert_checkpoints, wait_all_tasks_blocked
@@ -586,3 +587,31 @@ async def lock_taker() -> None:
         await wait_all_tasks_blocked()
         assert record == ["started"]
         lock_like.release()
+
+
+async def test_lock_acquire_unowned_lock() -> None:
+    """Test that trying to acquire a lock whose owner has exited raises an error.
+    Partial fix for https://github.com/python-trio/trio/issues/3035
+    """
+    lock = trio.Lock()
+    async with trio.open_nursery() as nursery:
+        nursery.start_soon(lock.acquire)
+    with pytest.raises(
+        trio.BrokenResourceError,
+        match="^Attempted acquire on Lock which will never be released, owner has exited.$",
+    ):
+        await lock.acquire()
+
+
+@slow
+async def test_lock_multiple_acquire() -> None:
+    """This currently hangs without throwing any error
+    See https://github.com/python-trio/trio/issues/3035
+    """
+    lock = trio.Lock()
+    with pytest.raises(trio.TooSlowError):
+        with trio.fail_after(1):
+            async with trio.open_nursery() as nursery:
+                nursery.start_soon(lock.acquire)
+                # this should probably raise trio.BrokenResourceError
+                nursery.start_soon(lock.acquire)