Wake tasks reparked into broken parking lots

newsfragments/3534.bugfix.rst

@@ -0,0 +1,3 @@
+Tasks moved into an already broken `trio.lowlevel.ParkingLot` by
+`~trio.lowlevel.ParkingLot.repark` or `~trio.lowlevel.ParkingLot.repark_all`
+now raise `trio.BrokenResourceError` instead of remaining blocked.

src/trio/_core/_parking_lot.py

@@ -256,6 +256,9 @@ async def main():
         If there are fewer than ``count`` tasks parked, then reparks as many
         tasks as are available and then returns successfully.
 
+        If ``new_lot`` is broken, the moved tasks' calls to :meth:`park`
+        raise :exc:`trio.BrokenResourceError` instead.
+
         Args:
           new_lot (ParkingLot): the parking lot to move tasks to.
           count (int|math.inf): the number of tasks to move.
@@ -264,8 +267,18 @@ async def main():
         if not isinstance(new_lot, ParkingLot):
             raise TypeError("new_lot must be a ParkingLot")
         for task in self._pop_several(count):
-            new_lot._parked[task] = None
-            task.custom_sleep_data = new_lot
+            if new_lot.broken_by:
+                _core.reschedule(
+                    task,
+                    outcome.Error(
+                        _core.BrokenResourceError(
+                            f"Attempted to repark in parking lot broken by {new_lot.broken_by}",
+                        ),
+                    ),
+                )
+            else:
+                new_lot._parked[task] = None
+                task.custom_sleep_data = new_lot
 
     def repark_all(self, new_lot: ParkingLot) -> None:
         """Move all parked tasks from one :class:`ParkingLot` object to

src/trio/_core/_tests/test_parking_lot.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import math
 import re
 from typing import TypeVar
 
@@ -225,6 +226,50 @@ async def test_parking_lot_repark_with_count() -> None:
         lot1.unpark_all()
 
 
+@pytest.mark.parametrize(
+    ("count", "expected_broken"),
+    [(0, 0), (1, 1), (2, 2), (5, 3), (math.inf, 3), (None, 3)],
+)
+async def test_parking_lot_repark_to_broken_lot(
+    count: float | None,
+    expected_broken: int,
+) -> None:
+    """Reparked tasks fail instead of waiting forever in a broken destination."""
+    source = ParkingLot()
+    target = ParkingLot()
+    target.break_lot()
+    broken: set[int] = set()
+    woken: set[int] = set()
+
+    async def waiter(i: int) -> None:
+        try:
+            await source.park()
+        except _core.BrokenResourceError:
+            broken.add(i)
+        else:
+            woken.add(i)
+
+    async with _core.open_nursery() as nursery:
+        for i in range(3):
+            nursery.start_soon(waiter, i)
+            await wait_all_tasks_blocked()
+
+        if count is None:
+            source.repark_all(target)
+        else:
+            source.repark(target, count=count)
+        await wait_all_tasks_blocked()
+
+        assert broken == set(range(expected_broken))
+        assert not woken
+        assert len(target) == 0
+        assert len(source) == 3 - expected_broken
+
+        source.unpark_all()
+
+    assert woken == set(range(expected_broken, 3))
+
+
 async def dummy_task(
     task_status: _core.TaskStatus[_core.Task] = trio.TASK_STATUS_IGNORED,
 ) -> None: