fix: distinguish shutdown cleanup outcomes from unexpected background failures

Supervisor treats early success/cancel as errors; cleanup only logs real
exceptions and skips already-handled outcomes so clean shutdown stays silent.
Assign the outcome marker directly to satisfy flake8 B010.

Author: songzhendong

skywalking/agent/__init__.py

@@ -147,25 +147,53 @@ async def _shutdown_async_queue(q: asyncio.Queue, label: str) -> None:
         )
 
 
-def _retrieve_background_task_outcome(task: asyncio.Task):
+def _retrieve_background_task_outcome(
+    task: asyncio.Task,
+    *,
+    report_unexpected_completion: bool = True,
+    report_unexpected_cancellation: bool = False,
+):
     """
-    Return a completed background task's exception, or a sentinel for unexpected success.
+    Return a completed background task's exception when it should be reported.
 
     Retrieves the exception so asyncio does not emit "never retrieved" warnings.
+    Successful completion / cancellation are only converted to errors when the
+    corresponding report_* flag is set (supervisor path before shutdown).
     """
-    if task is None or not task.done() or task.cancelled():
+    if task is None or not task.done():
+        return None
+    if task.cancelled():
+        if report_unexpected_cancellation:
+            return RuntimeError(
+                'Python agent asyncio background task was cancelled unexpectedly'
+            )
         return None
     exc = task.exception()
     if exc is not None:
         return exc
-    return RuntimeError('Python agent asyncio background task finished unexpectedly')
+    if report_unexpected_completion:
+        return RuntimeError('Python agent asyncio background task finished unexpectedly')
+    return None
 
 
-def _log_background_task_outcome(task: asyncio.Task) -> bool:
+def _log_background_task_outcome(
+    task: asyncio.Task,
+    *,
+    report_unexpected_completion: bool = True,
+    report_unexpected_cancellation: bool = False,
+) -> bool:
     """Log and retrieve a completed background task outcome. Returns True if logged."""
-    exc = _retrieve_background_task_outcome(task)
+    if getattr(task, '_sw_outcome_handled', False):
+        return False
+    exc = _retrieve_background_task_outcome(
+        task,
+        report_unexpected_completion=report_unexpected_completion,
+        report_unexpected_cancellation=report_unexpected_cancellation,
+    )
     if exc is None:
         return False
+    # Mark before logging so cleanup never re-logs a supervisor-handled outcome.
+    task._sw_outcome_handled = True
     logger.error('Error in Python agent asyncio event loop: %s', exc, exc_info=exc)
     return True
 
@@ -177,8 +205,9 @@ async def _await_shutdown_or_background_failure(
     """
     Wait for shutdown or the first unexpected background-task completion.
 
-    Background tasks are expected to run until shutdown. If one finishes early,
-    retrieve/log its outcome and signal shutdown so the root can clean up.
+    Background tasks are expected to run until shutdown. If one finishes or is
+    cancelled early, retrieve/log its outcome and signal shutdown so the root
+    can clean up.
     """
     shutdown_waiter = asyncio.create_task(finished.wait())
     pending = {task for task in background_tasks if task is not None}
@@ -195,7 +224,12 @@ async def _await_shutdown_or_background_failure(
                 return
             for task in done:
                 pending.discard(task)
-                if _log_background_task_outcome(task):
+                # Before shutdown, both early success and unexpected cancel are errors.
+                if _log_background_task_outcome(
+                    task,
+                    report_unexpected_completion=True,
+                    report_unexpected_cancellation=True,
+                ):
                     finished.set()
                     return
     finally:
@@ -213,13 +247,20 @@ async def _cancel_pending_tasks(tasks) -> None:
 
     Cancel only the given reporter / watch tasks; never the asyncio.run root.
     The current task is excluded so we never await ourselves.
+
+    During cleanup, normal completion and intentional cancellation are silent;
+    only real exceptions that were not already handled by the supervisor are logged.
     """
     current = asyncio.current_task()
     for task in tasks:
         if task is None or task is current:
             continue
         if task.done():
-            _log_background_task_outcome(task)
+            _log_background_task_outcome(
+                task,
+                report_unexpected_completion=False,
+                report_unexpected_cancellation=False,
+            )
     pending = [
         task for task in tasks
         if task is not None and task is not current and not task.done()
@@ -240,7 +281,11 @@ async def _cancel_pending_tasks(tasks) -> None:
             sum(1 for task in pending if not task.done()),
         )
     for task in pending:
-        _log_background_task_outcome(task)
+        _log_background_task_outcome(
+            task,
+            report_unexpected_completion=False,
+            report_unexpected_cancellation=False,
+        )
 
 
 def _close_previous_protocol(protocol) -> None:

tests/unit/test_shutdown_queue.py

@@ -179,14 +179,16 @@ async def reporter():
         self.assertTrue(holder.get('root_finished', False))
 
     def test_background_task_failure_is_logged_not_silenced(self):
-        """Regression: failing background tasks must be observed and logged."""
-        holder = {}
+        """Regression: failing background tasks must be observed and logged exactly once."""
+        holder = {'errors': []}
         error_logged = Event()
+        aclose_done = Event()
 
         class _Handler(logging.Handler):
             def emit(self, record):
-                if 'Error in Python agent asyncio event loop' in record.getMessage():
-                    holder['logged'] = record.getMessage()
+                msg = record.getMessage()
+                if 'Error in Python agent asyncio event loop' in msg:
+                    holder['errors'].append(msg)
                     error_logged.set()
 
         agent_logger = logging.getLogger('skywalking')
@@ -199,6 +201,10 @@ async def failing_background():
             await asyncio.sleep(0.02)
             raise ValueError('command dispatch failed')
 
+        async def yielding_aclose():
+            await asyncio.sleep(0.02)
+            aclose_done.set()
+
         async def root():
             finished = asyncio.Event()
             holder['finished'] = finished
@@ -211,21 +217,150 @@ async def reporter():
             reporter_task = asyncio.create_task(reporter())
             tasks = {failing_task, reporter_task}
             await _await_shutdown_or_background_failure(finished, tasks)
+            await _cancel_pending_tasks(tasks)
+            await yielding_aclose()
             holder['after_wait'] = True
             holder['failing_task'] = failing_task
 
         try:
             asyncio.run(root())
             self.assertTrue(holder.get('after_wait'))
             self.assertTrue(error_logged.wait(2.0))
-            self.assertIn('command dispatch failed', holder.get('logged', ''))
+            self.assertTrue(aclose_done.wait(2.0))
+            self.assertEqual(len(holder['errors']), 1)
+            self.assertIn('command dispatch failed', holder['errors'][0])
             self.assertTrue(holder['finished'].is_set())
             self.assertTrue(holder['failing_task'].done())
             self.assertIsInstance(holder['failing_task'].exception(), ValueError)
         finally:
             agent_logger.removeHandler(handler)
             agent_logger.setLevel(previous_level)
 
+    def test_clean_shutdown_does_not_log_normal_background_completion(self):
+        """Clean shutdown: reporters finish after _finished; cleanup must stay silent."""
+        holder = {'errors': []}
+
+        class _Handler(logging.Handler):
+            def emit(self, record):
+                if 'Error in Python agent asyncio event loop' in record.getMessage():
+                    holder['errors'].append(record.getMessage())
+
+        agent_logger = logging.getLogger('skywalking')
+        handler = _Handler()
+        agent_logger.addHandler(handler)
+        previous_level = agent_logger.level
+        agent_logger.setLevel(logging.ERROR)
+
+        async def root():
+            finished = asyncio.Event()
+
+            async def reporter():
+                while not finished.is_set():
+                    await asyncio.sleep(0.01)
+
+            tasks = {asyncio.create_task(reporter()) for _ in range(2)}
+            finished.set()
+            await _await_shutdown_or_background_failure(finished, tasks)
+            # Give reporters a turn to observe _finished and return normally.
+            await asyncio.sleep(0.05)
+            await _cancel_pending_tasks(tasks)
+            holder['all_done'] = all(task.done() for task in tasks)
+
+        try:
+            asyncio.run(root())
+            self.assertTrue(holder.get('all_done'))
+            self.assertEqual(holder['errors'], [])
+        finally:
+            agent_logger.removeHandler(handler)
+            agent_logger.setLevel(previous_level)
+
+    def test_unexpected_pre_shutdown_cancellation_is_logged(self):
+        """Cancellation before shutdown must be logged and trigger orderly cleanup."""
+        holder = {'errors': []}
+        error_logged = Event()
+        aclose_done = Event()
+
+        class _Handler(logging.Handler):
+            def emit(self, record):
+                msg = record.getMessage()
+                if 'Error in Python agent asyncio event loop' in msg:
+                    holder['errors'].append(msg)
+                    error_logged.set()
+
+        agent_logger = logging.getLogger('skywalking')
+        handler = _Handler()
+        agent_logger.addHandler(handler)
+        previous_level = agent_logger.level
+        agent_logger.setLevel(logging.ERROR)
+
+        async def cancelled_background():
+            asyncio.current_task().cancel()
+            await asyncio.sleep(0)
+
+        async def yielding_aclose():
+            await asyncio.sleep(0.02)
+            aclose_done.set()
+
+        async def root():
+            finished = asyncio.Event()
+            holder['finished'] = finished
+
+            async def reporter():
+                while not finished.is_set():
+                    await asyncio.sleep(0.05)
+
+            cancelled_task = asyncio.create_task(cancelled_background())
+            reporter_task = asyncio.create_task(reporter())
+            tasks = {cancelled_task, reporter_task}
+            await _await_shutdown_or_background_failure(finished, tasks)
+            await _cancel_pending_tasks(tasks)
+            await yielding_aclose()
+            holder['after_wait'] = True
+
+        try:
+            asyncio.run(root())
+            self.assertTrue(holder.get('after_wait'))
+            self.assertTrue(error_logged.wait(2.0))
+            self.assertTrue(aclose_done.wait(2.0))
+            self.assertEqual(len(holder['errors']), 1)
+            self.assertIn('cancelled unexpectedly', holder['errors'][0])
+            self.assertTrue(holder['finished'].is_set())
+        finally:
+            agent_logger.removeHandler(handler)
+            agent_logger.setLevel(previous_level)
+
+    def test_intentional_cleanup_cancellation_is_silent(self):
+        """Tasks cancelled by _cancel_pending_tasks during cleanup must not ERROR."""
+        holder = {'errors': []}
+
+        class _Handler(logging.Handler):
+            def emit(self, record):
+                if 'Error in Python agent asyncio event loop' in record.getMessage():
+                    holder['errors'].append(record.getMessage())
+
+        agent_logger = logging.getLogger('skywalking')
+        handler = _Handler()
+        agent_logger.addHandler(handler)
+        previous_level = agent_logger.level
+        agent_logger.setLevel(logging.ERROR)
+
+        async def forever():
+            while True:
+                await asyncio.sleep(0.05)
+
+        async def root():
+            tasks = {asyncio.create_task(forever()) for _ in range(2)}
+            await _cancel_pending_tasks(tasks)
+            holder['all_done'] = all(task.done() for task in tasks)
+
+        try:
+            asyncio.run(root())
+            self.assertTrue(holder.get('all_done'))
+            self.assertEqual(holder['errors'], [])
+        finally:
+            agent_logger.removeHandler(handler)
+            agent_logger.setLevel(previous_level)
+
     def test_shutdown_async_queue_bounded(self):
         async def _run():
             q = asyncio.Queue()