fix: Address Bugbot feedback — stale terminator and flush components

- AsyncWorker.kill(): Reset queue to None instead of putting a stale
  _TERMINATOR (since we now cancel the task directly, the terminator
  was never consumed and would break restart)
- close() with async transport: Call _flush_components() to flush
  session flusher, log/metrics/span batchers even when sync flush
  is skipped
- Update test to verify fresh queue creation after kill

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

Author: BYK

sentry_sdk/client.py

@@ -1045,6 +1045,7 @@ def close(
                     "Prefer close_async() for graceful async shutdown. "
                     "Performing synchronous best-effort cleanup."
                 )
+                self._flush_components()
             else:
                 self.flush(timeout=timeout, callback=callback)
             self._close_components()

sentry_sdk/worker.py

@@ -213,17 +213,14 @@ def kill(self) -> None:
         if self._task:
             # Cancel the main consumer task to prevent duplicate consumers
             self._task.cancel()
-            if self._queue is not 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()
+            # Reset queue to avoid stale terminators on restart
+            self._queue = None
             self._loop = None
             self._task = None
             self._task_for_pid = None

tests/test_transport.py

@@ -1028,24 +1028,27 @@ async def test_async_transport_background_thread_capture(
     client = make_client(_experiments=experiments, integrations=[AsyncioIntegration()])
     assert isinstance(client.transport, AsyncHttpTransport)
     sentry_sdk.get_global_scope().set_client(client)
-    captured_from_thread = []
-    exception_from_thread = []
-
-    def background_thread_work():
-        try:
-            # This should use run_coroutine_threadsafe path
-            capture_message("from background thread")
-            captured_from_thread.append(True)
-        except Exception as e:
-            exception_from_thread.append(e)
-
-    thread = threading.Thread(target=background_thread_work)
-    thread.start()
-    thread.join()
-    assert not exception_from_thread
-    assert captured_from_thread
-    await client.close_async(timeout=2.0)
-    assert capturing_server.captured
+    try:
+        captured_from_thread = []
+        exception_from_thread = []
+
+        def background_thread_work():
+            try:
+                # This should use run_coroutine_threadsafe path
+                capture_message("from background thread")
+                captured_from_thread.append(True)
+            except Exception as e:
+                exception_from_thread.append(e)
+
+        thread = threading.Thread(target=background_thread_work)
+        thread.start()
+        thread.join()
+        assert not exception_from_thread
+        assert captured_from_thread
+        await client.close_async(timeout=2.0)
+        assert capturing_server.captured
+    finally:
+        sentry_sdk.get_global_scope().set_client(None)
 
 
 @skip_under_gevent
@@ -1276,18 +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_reuses_existing_queue():
-    """Test start() reuses existing queue if already created."""
+async def test_async_worker_start_creates_fresh_queue_after_kill():
+    """Test start() creates a fresh queue after kill() resets it."""
     from sentry_sdk.worker import AsyncWorker
 
     worker = AsyncWorker(queue_size=10)
     worker.start()
-    queue_ref = worker._queue
-    # Kill and restart — queue should be reused
+    assert worker._queue is not None
+    # Kill resets queue to None to avoid stale terminators
     worker.kill()
     await asyncio.sleep(0)  # Allow cancelled tasks to be cleaned up
+    assert worker._queue is None
+    # Restart creates a fresh queue
     worker.start()
-    assert worker._queue is queue_ref
+    assert worker._queue is not None
     worker.kill()
     await asyncio.sleep(0)  # Allow cancelled tasks to be cleaned up