Why is the "the change stream loop is never exited"? If you don't respond to comments that require a response, you are only making it more difficult for people to help you. Also, if you receive one or more answers that resolves your question, you should select the best answer and accept it (see What should I do when someone answers my question?). When you have increased your reputation, you can also upvote one or more questions that are useful.

If you are running Python 3.12 or later, you can do:

async def handle_collection_changes(...):
    loop = asyncio.get_running_loop
    loop.set_task_factory(asyncio.eager_task_factory)

See asyncio.eager_task_factory. Now when you execute task = asyncio.create_task(coro.), coro will be immediately invoked. The new task will not be added to the event loop (it's as if a synchronous call were being made to coro) and run until either it completes or until it issues a blocking await expression in which case it will then be added to the event loop like any regular task. Setting this factory will of course result in all tasks being created "eagerly". This is generally not a problem, but if it is then instantiate the tasks that you want to start "eagerly" with:

loop = asyncio.get_running_loop()
task = asyncio.Task(coro, loop=loop, eager_start=True)