Remove logcontext problems caused by awaiting raw deferLater(...) (#19058)

This is a normal
problem where we `await` a deferred without wrapping it in
`make_deferred_yieldable(...)`. But I've opted to replace the usage of
`deferLater` with something more standard for the Synapse codebase.

Part of #18905

It's unclear why we're only now seeing these failures happen with the
changes from #19057

Example failures seen in
https://github.com/element-hq/synapse/actions/runs/18477454390/job/52645183606?pr=19057

```
builtins.AssertionError: Expected `looping_call` callback from the reactor to start with the sentinel logcontext but saw task-_resumable_task-0-IBzAmHUoepQfLnEA. In other words, another task shouldn't have leaked their logcontext to us.
```

Author: MadLittleMods

changelog.d/19058.misc

@@ -0,0 +1 @@
+Remove logcontext problems caused by awaiting raw `deferLater(...)`.

synapse/util/task_scheduler.py

@@ -53,8 +53,8 @@
 class TaskScheduler:
     """
     This is a simple task scheduler designed for resumable tasks. Normally,
-    you'd use `run_in_background` to start a background task or Twisted's
-    `deferLater` if you want to run it later.
+    you'd use `run_in_background` to start a background task or `clock.call_later`
+    if you want to run it later.
 
     The issue is that these tasks stop completely and won't resume if Synapse is
     shut down for any reason.

tests/rest/admin/test_room.py

@@ -27,7 +27,6 @@
 
 from parameterized import parameterized
 
-from twisted.internet.task import deferLater
 from twisted.internet.testing import MemoryReactor
 
 import synapse.rest.admin
@@ -1163,7 +1162,7 @@ def test_delete_same_room_twice(self) -> None:
         # Mock PaginationHandler.purge_room to sleep for 100s, so we have time to do a second call
         # before the purge is over. Note that it doesn't purge anymore, but we don't care.
         async def purge_room(room_id: str, force: bool) -> None:
-            await deferLater(self.hs.get_reactor(), 100, lambda: None)
+            await self.hs.get_clock().sleep(100)
 
         self.pagination_handler.purge_room = AsyncMock(side_effect=purge_room)  # type: ignore[method-assign]
 

tests/util/test_task_scheduler.py

@@ -20,9 +20,10 @@
 #
 from typing import Optional
 
-from twisted.internet.task import deferLater
+from twisted.internet.defer import Deferred
 from twisted.internet.testing import MemoryReactor
 
+from synapse.logging.context import make_deferred_yieldable
 from synapse.server import HomeServer
 from synapse.types import JsonMapping, ScheduledTask, TaskStatus
 from synapse.util.clock import Clock
@@ -87,7 +88,7 @@ async def _sleeping_task(
         self, task: ScheduledTask
     ) -> tuple[TaskStatus, Optional[JsonMapping], Optional[str]]:
         # Sleep for a second
-        await deferLater(self.reactor, 1, lambda: None)
+        await self.hs.get_clock().sleep(1)
         return TaskStatus.COMPLETE, None, None
 
     def test_schedule_lot_of_tasks(self) -> None:
@@ -170,8 +171,10 @@ async def _resumable_task(
             return TaskStatus.COMPLETE, {"success": True}, None
         else:
             await self.task_scheduler.update_task(task.id, result={"in_progress": True})
+            # Create a deferred which we will never complete
+            incomplete_d: Deferred = Deferred()
             # Await forever to simulate an aborted task because of a restart
-            await deferLater(self.reactor, 2**16, lambda: None)
+            await make_deferred_yieldable(incomplete_d)
             # This should never been called
             return TaskStatus.ACTIVE, None, None