fix: Address bot feedback from merge

- AsyncWorker: Create fresh queue on start() instead of nullifying in
  kill(). This avoids the race where kill() nulls the queue before
  _on_task_complete can call task_done(), which would hang queue.join().
- AsyncHttpTransport._get_pool_options: Respect keep_alive option
  (was unconditionally adding keep-alive socket options).
- close() with async transport: Call _flush_components() before cleanup.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: BYK

sentry_sdk/transport.py

@@ -891,10 +891,11 @@ def _get_pool_options(self: "Self") -> "Dict[str, Any]":
                 else []
             )
 
-            used_options = {(o[0], o[1]) for o in socket_options}
-            for default_option in KEEP_ALIVE_SOCKET_OPTIONS:
-                if (default_option[0], default_option[1]) not in used_options:
-                    socket_options.append(default_option)
+            if self.options["keep_alive"]:
+                used_options = {(o[0], o[1]) for o in socket_options}
+                for default_option in KEEP_ALIVE_SOCKET_OPTIONS:
+                    if (default_option[0], default_option[1]) not in used_options:
+                        socket_options.append(default_option)
 
             options["socket_options"] = socket_options
 

sentry_sdk/worker.py

@@ -219,8 +219,6 @@ def kill(self) -> None:
             for task in tasks_to_cancel:
                 task.cancel()
             self._active_tasks.clear()
-            # Reset queue to avoid stale terminators on restart
-            self._queue = None
             self._loop = None
             self._task = None
             self._task_for_pid = None
@@ -229,8 +227,8 @@ def start(self) -> None:
         if not self.is_alive:
             try:
                 self._loop = asyncio.get_running_loop()
-                if self._queue is None:
-                    self._queue = asyncio.Queue(maxsize=self._queue_size)
+                # Always create a fresh queue on start to avoid stale items
+                self._queue = asyncio.Queue(maxsize=self._queue_size)
                 with mark_sentry_task_internal():
                     self._task = self._loop.create_task(self._target())
                 self._task_for_pid = os.getpid()

tests/test_transport.py

@@ -1279,20 +1279,20 @@ def test_async_worker_start_no_running_loop():
 @pytest.mark.asyncio
 @pytest.mark.skipif(not PY38, reason="AsyncWorker requires Python 3.8+")
 @pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning")
-async def test_async_worker_start_creates_fresh_queue_after_kill():
-    """Test start() creates a fresh queue after kill() resets it."""
+async def test_async_worker_start_creates_fresh_queue_on_restart():
+    """Test start() creates a fresh queue on each restart."""
     from sentry_sdk.worker import AsyncWorker
 
     worker = AsyncWorker(queue_size=10)
     worker.start()
-    assert worker._queue is not None
-    # Kill resets queue to None to avoid stale terminators
+    old_queue = worker._queue
+    assert old_queue is not None
     worker.kill()
     await asyncio.sleep(0)  # Allow cancelled tasks to be cleaned up
-    assert worker._queue is None
-    # Restart creates a fresh queue
+    # Restart creates a fresh queue (avoids stale items)
     worker.start()
     assert worker._queue is not None
+    assert worker._queue is not old_queue
     worker.kill()
     await asyncio.sleep(0)  # Allow cancelled tasks to be cleaned up