Demonstrate a potential deadlock with service nurseries

service_nursery_deadlock.py

import trio
import pytest
from contextlib import asynccontextmanager
from service_nursery import run_service_nursery


@asynccontextmanager
async def run_deadlock_service():

    class service:
        pass

    async def target():
        try:
            service.running = True
            service.stopped = trio.Event()
            await trio.sleep_forever()
        finally:
            with trio.CancelScope(shield=True):
                print("Cleanup target")
                service.running = False
                await trio.sleep(.2)
                service.stopped.set()

    async with open_service_nursery() as nursery:
        nursery.start_soon(target)
        try:
            yield service
        finally:
            with trio.CancelScope(shield=True):
                print("Cleanup with target running")
                await trio.sleep(.1)
                assert service.running

                print("Cancel and join target")
                # Uncomment this line to fix the deadlock
                # nursery.child_task_scopes.shield = False
                nursery.cancel_scope.cancel()
                await service.stopped.wait()
                
                print("Cleanup with target stopped")
                assert not service.running
                await trio.sleep(.1)


@pytest.mark.trio
async def test_deadlock():
    async with run_deadlock_service():
        await trio.sleep(.1)