fix(worker): Modify kill behaviour to mirror threaded worker

Changed kill to also use the _TERMINATOR sentinel, so the queue is still drained to this point on kill instead of cancelled immediately. This should also fix potential race conditions with flush_async.

GH-4581

Author: srothh

sentry_sdk/worker.py

@@ -192,7 +192,7 @@ def _target(self) -> None:
 
 class AsyncWorker(Worker):
     def __init__(self, queue_size: int = DEFAULT_QUEUE_SIZE) -> None:
-        self._queue: asyncio.Queue[Callable[[], Any]] = asyncio.Queue(queue_size)
+        self._queue: asyncio.Queue[Any] = asyncio.Queue(queue_size)
         self._task: Optional[asyncio.Task[None]] = None
         # Event loop needs to remain in the same process
         self._task_for_pid: Optional[int] = None
@@ -210,16 +210,19 @@ def is_alive(self) -> bool:
 
     def kill(self) -> None:
         if self._task:
-            self._task.cancel()
-            self._task = None
-            self._task_for_pid = None
+            try:
+                self._queue.put_nowait(_TERMINATOR)
+            except asyncio.QueueFull:
+                logger.debug("async worker queue full, kill failed")
             # Also cancel any active callback tasks
             # Avoid modifying the set while cancelling tasks
             tasks_to_cancel = set(self._active_tasks)
             for task in tasks_to_cancel:
                 task.cancel()
             self._active_tasks.clear()
             self._loop = None
+            self._task = None
+            self._task_for_pid = None
 
     def start(self) -> None:
         if not self.is_alive:
@@ -280,6 +283,9 @@ def submit(self, callback: Callable[[], Any]) -> bool:
     async def _target(self) -> None:
         while True:
             callback = await self._queue.get()
+            if callback is _TERMINATOR:
+                self._queue.task_done()
+                break
             # Firing tasks instead of awaiting them allows for concurrent requests
             task = asyncio.create_task(self._process_callback(callback))
             # Create a strong reference to the task so it can be cancelled on kill