fix: Capture queue ref at dispatch time in _on_task_complete

Bind the queue reference when the task is dispatched, not when the
done callback fires. This prevents kill()/start() from replacing
self._queue before old callbacks can call task_done(), which would
corrupt the new queue's unfinished_tasks counter.

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

Author: BYK

sentry_sdk/worker.py

@@ -305,7 +305,10 @@ async def _target(self) -> None:
                 # Create a strong reference to the task so it can be cancelled on kill
                 # and does not get garbage collected while running
                 self._active_tasks.add(task)
-                task.add_done_callback(self._on_task_complete)
+                # Capture queue ref at dispatch time so done callbacks use the
+                # correct queue even if kill()/start() replace self._queue.
+                queue_ref = self._queue
+                task.add_done_callback(lambda t: self._on_task_complete(t, queue_ref))
                 # Yield to let the event loop run other tasks
                 await asyncio.sleep(0)
         except asyncio.CancelledError:
@@ -315,7 +318,11 @@ async def _process_callback(self, callback: "Callable[[], Any]") -> None:
         # Callback is an async coroutine, need to await it
         await callback()
 
-    def _on_task_complete(self, task: "asyncio.Task[None]") -> None:
+    def _on_task_complete(
+        self,
+        task: "asyncio.Task[None]",
+        queue: "Optional[asyncio.Queue[Any]]" = None,
+    ) -> None:
         try:
             task.result()
         except asyncio.CancelledError:
@@ -324,7 +331,8 @@ def _on_task_complete(self, task: "asyncio.Task[None]") -> None:
             logger.error("Failed processing job", exc_info=True)
         finally:
             # Mark the task as done and remove it from the active tasks set
-            # This happens only after the task has completed
-            if self._queue is not None:
-                self._queue.task_done()
+            # Use the queue reference captured at dispatch time, not self._queue,
+            # to avoid calling task_done() on a different queue after kill()/start().
+            if queue is not None:
+                queue.task_done()
             self._active_tasks.discard(task)

tests/test_transport.py

@@ -1656,7 +1656,7 @@ async def will_be_cancelled():
 
     # _on_task_complete should handle CancelledError without logging error
     with mock.patch("sentry_sdk.worker.logger") as mock_logger:
-        worker._on_task_complete(task)
+        worker._on_task_complete(task, worker._queue)
         mock_logger.error.assert_not_called()
 
     # Verify task was discarded from active_tasks
@@ -1716,7 +1716,7 @@ async def simple_cb():
     mock_task.result.return_value = None
     worker._active_tasks.add(mock_task)
 
-    worker._on_task_complete(mock_task)
+    worker._on_task_complete(mock_task, worker._queue)
     assert mock_task not in worker._active_tasks
     worker.kill()
     await asyncio.sleep(0)  # Allow cancelled tasks to be cleaned up
@@ -2864,7 +2864,7 @@ def test_sync_cov_async_worker_on_task_complete_success():
     mock_task.result.return_value = None
     worker._active_tasks.add(mock_task)
 
-    worker._on_task_complete(mock_task)
+    worker._on_task_complete(mock_task, worker._queue)
 
     mock_task.result.assert_called_once()
     mock_queue.task_done.assert_called_once()
@@ -2885,7 +2885,7 @@ def test_sync_cov_async_worker_on_task_complete_cancelled():
     worker._active_tasks.add(mock_task)
 
     with mock.patch("sentry_sdk.worker.logger") as mock_logger:
-        worker._on_task_complete(mock_task)
+        worker._on_task_complete(mock_task, worker._queue)
         # CancelledError should NOT trigger error logging
         mock_logger.error.assert_not_called()
 
@@ -2907,7 +2907,7 @@ def test_sync_cov_async_worker_on_task_complete_exception():
     worker._active_tasks.add(mock_task)
 
     with mock.patch("sentry_sdk.worker.logger") as mock_logger:
-        worker._on_task_complete(mock_task)
+        worker._on_task_complete(mock_task, worker._queue)
         mock_logger.error.assert_called_once_with(
             "Failed processing job", exc_info=True
         )
@@ -2929,7 +2929,7 @@ def test_sync_cov_async_worker_on_task_complete_queue_none():
     worker._active_tasks.add(mock_task)
 
     # Should not raise despite queue being None
-    worker._on_task_complete(mock_task)
+    worker._on_task_complete(mock_task, worker._queue)
     assert mock_task not in worker._active_tasks