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

@@ -1276,18 +1276,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