♻️Maintenance: improve cancellation error handling (#8367)

Author: sanderegg

.github/instructions/general.instructions.md

@@ -0,0 +1,11 @@
+---
+applyTo: '**'
+---
+Provide project context and coding guidelines that AI should follow when generating code, answering questions, or reviewing changes.
+
+## General Guidelines
+
+1. **Test-Driven Development**: Write unit tests for all new functions and features. Use `pytest` for Python and appropriate testing frameworks for Node.js.
+2. **Environment Variables**: Use [Environment Variables Guide](../../docs/env-vars.md) for configuration. Avoid hardcoding sensitive information.
+3. **Documentation**: Prefer self-explanatory code; add documentation only if explicitly requested by the developer. Be concise.
+4. **Code Reviews**: Participate in code reviews and provide constructive feedback.

.github/instructions/node.instructions.md

@@ -0,0 +1,11 @@
+---
+applyTo: '**/*.js'
+---
+Provide project context and coding guidelines that AI should follow when generating code, answering questions, or reviewing changes.
+
+## 🛠️Coding Instructions for Node.js in This Repository
+
+* Use ES6+ syntax and features.
+* Follow the `package.json` configuration for dependencies and scripts.
+* Use `eslint` for linting and `prettier` for code formatting.
+* Write modular and reusable code, adhering to the project's structure.

.github/copilot-instructions.md renamed to .github/instructions/python.instructions.md

@@ -1,14 +1,7 @@
-# GitHub Copilot Instructions
-
-This document provides guidelines and best practices for using GitHub Copilot in the `osparc-simcore` repository and other Python and Node.js projects.
-
-## General Guidelines
-
-1. **Test-Driven Development**: Write unit tests for all new functions and features. Use `pytest` for Python and appropriate testing frameworks for Node.js.
-2. **Environment Variables**: Use [Environment Variables Guide](../docs/env-vars.md) for configuration. Avoid hardcoding sensitive information.
-3. **Documentation**: Prefer self-explanatory code; add documentation only if explicitly requested by the developer.
-
 ---
+applyTo: '**/*.py'
+---
+Provide project context and coding guidelines that AI should follow when generating code, answering questions, or reviewing changes.
 
 ## 🛠️Coding Instructions for Python in This Repository
 
@@ -25,8 +18,8 @@ Follow these rules **strictly** when generating Python code:
 
 ### 3. **Code Style & Formatting**
 
-* Follow [Python Coding Conventions](../docs/coding-conventions.md) **strictly**.
-* Format code with `black`.
+* Follow [Python Coding Conventions](../../docs/coding-conventions.md) **strictly**.
+* Format code with `black` and `ruff`.
 * Lint code with `ruff` and `pylint`.
 
 ### 4. **Library Compatibility**
@@ -51,11 +44,6 @@ Ensure compatibility with the following library versions:
 * Prefer `json_dumps` / `json_loads` from `common_library.json_serialization` instead of the built-in `json.dumps` / `json.loads`.
 * When using Pydantic models, prefer methods like `model.model_dump_json()` for serialization.
 
----
-
-## 🛠️Coding Instructions for Node.js in This Repository
-
-* Use ES6+ syntax and features.
-* Follow the `package.json` configuration for dependencies and scripts.
-* Use `eslint` for linting and `prettier` for code formatting.
-* Write modular and reusable code, adhering to the project's structure.
+### 7. **Running tests**
+* Use `--keep-docker-up` flag when testing to keep docker containers up between sessions.
+* Always activate the python virtual environment before running pytest.

packages/common-library/src/common_library/async_tools.py

@@ -89,32 +89,37 @@ async def cancel_wait_task(
         TimeoutError: raised if cannot cancel the task.
         CancelledError: raised ONLY if owner is being cancelled.
     """
+    if task.done():
+        # nothing to do here
+        return
 
-    cancelling = task.cancel()
-    if not cancelling:
-        return  # task was alredy cancelled
-
-    assert task.cancelling()  # nosec
-    assert not task.cancelled()  # nosec
-
+    # mark for cancellation
+    task.cancel("cancel_wait_task was called to cancel this task")
     try:
-
+        _logger.debug("Cancelling task %s", task.get_name())
         await asyncio.shield(
             # NOTE shield ensures that cancellation of the caller function won't stop you
             # from observing the cancellation/finalization of task.
             asyncio.wait_for(task, timeout=max_delay)
         )
-
-    except asyncio.CancelledError:
-        if not task.cancelled():
-            # task owner function is being cancelled -> propagate cancellation
-            raise
-
-        # else: task cancellation is complete, we can safely ignore it
-        _logger.debug(
-            "Task %s cancellation is complete",
+    except TimeoutError:
+        _logger.exception(
+            "Timeout while cancelling task %s after %s seconds",
             task.get_name(),
+            max_delay,
         )
+        raise
+    except asyncio.CancelledError:
+        current_task = asyncio.current_task()
+        assert current_task is not None  # nosec
+        if current_task.cancelling() > 0:
+            # owner function is being cancelled -> propagate cancellation
+            raise
+    finally:
+        if not task.done():
+            _logger.error("Failed to cancel %s", task.get_name())
+        else:
+            _logger.debug("Task %s cancelled", task.get_name())
 
 
 def delayed_start(

packages/service-library/src/servicelib/async_utils.py

@@ -2,11 +2,12 @@
 import logging
 from collections import deque
 from collections.abc import Awaitable, Callable
-from contextlib import suppress
 from dataclasses import dataclass
 from functools import wraps
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
+from common_library.async_tools import cancel_wait_task
+
 from . import tracing
 from .utils_profiling_middleware import dont_profile, is_profiling, profile_context
 
@@ -54,9 +55,7 @@ async def _safe_cancel(context: Context) -> None:
     try:
         await context.in_queue.put(None)
         if context.task is not None:
-            context.task.cancel()
-            with suppress(asyncio.CancelledError):
-                await context.task
+            await cancel_wait_task(context.task, max_delay=None)
     except RuntimeError as e:
         if "Event loop is closed" in f"{e}":
             _logger.warning("event loop is closed and could not cancel %s", context)

packages/service-library/src/servicelib/deferred_tasks/_worker_tracker.py

@@ -57,6 +57,12 @@ async def handle_run(
             result_to_return = TaskResultSuccess(value=task_result)
         except asyncio.CancelledError:
             result_to_return = TaskResultCancelledError()
+            # NOTE: if the task is itself cancelled it shall re-raise: see https://superfastpython.com/asyncio-cancellederror-consumed/
+            current_task = asyncio.current_task()
+            assert current_task is not None  # nosec
+            if current_task.cancelling() > 0:
+                # owner function is being cancelled -> propagate cancellation
+                raise
         except Exception as e:  # pylint:disable=broad-exception-caught
             result_to_return = TaskResultError(
                 error=_format_exception(e),

packages/service-library/src/servicelib/fastapi/requests_decorators.py

@@ -4,6 +4,7 @@
 from functools import wraps
 from typing import Any, Protocol
 
+from common_library.async_tools import cancel_wait_task
 from fastapi import Request, status
 from fastapi.exceptions import HTTPException
 
@@ -13,8 +14,7 @@
 class _HandlerWithRequestArg(Protocol):
     __name__: str
 
-    async def __call__(self, request: Request, *args: Any, **kwargs: Any) -> Any:
-        ...
+    async def __call__(self, request: Request, *args: Any, **kwargs: Any) -> Any: ...
 
 
 def _validate_signature(handler: _HandlerWithRequestArg):
@@ -75,13 +75,8 @@ async def wrapper(request: Request, *args, **kwargs):
 
         # One has completed, cancel the other
         for t in pending:
-            t.cancel()
-
             try:
-                await asyncio.wait_for(t, timeout=3)
-
-            except asyncio.CancelledError:
-                pass
+                await cancel_wait_task(t, max_delay=3)
             except Exception:  # pylint: disable=broad-except
                 if t is handler_task:
                     raise

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -335,6 +335,12 @@ async def _tasks_monitor(self) -> None:
                     result_field = ResultField(
                         str_error=dumps(TaskCancelledError(task_id=task_id))
                     )
+                    # NOTE: if the task is itself cancelled it shall re-raise: see https://superfastpython.com/asyncio-cancellederror-consumed/
+                    current_task = asyncio.current_task()
+                    assert current_task is not None  # nosec
+                    if current_task.cancelling() > 0:
+                        # owner function is being cancelled -> propagate cancellation
+                        raise
                 except Exception as e:  # pylint:disable=broad-except
                     allowed_errors = TaskRegistry.get_allowed_errors(
                         task_data.registered_task_name

packages/service-library/src/servicelib/redis/_decorators.py

@@ -8,6 +8,7 @@
 
 import arrow
 import redis.exceptions
+from common_library.async_tools import cancel_wait_task
 from redis.asyncio.lock import Lock
 
 from ..background_task import periodic
@@ -116,10 +117,12 @@ async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                             module_name=coro.__module__, func_name=coro.__name__
                         ),
                     )
-
                     res = await work_task
-                    auto_extend_lock_task.cancel()
-                    return res
+                    # cancel the auto-extend task (work is done)
+                    # NOTE: if we do not explicitely await the task inside the context manager
+                    # it sometimes hangs forever (Python issue?)
+                    await cancel_wait_task(auto_extend_lock_task, max_delay=None)
+                return res
 
             except BaseExceptionGroup as eg:
                 # Separate exceptions into LockLostError and others

packages/service-library/src/servicelib/redis/_utils.py

@@ -1,6 +1,6 @@
 import logging
 from collections.abc import Awaitable
-from typing import Any
+from typing import ParamSpec, TypeVar
 
 import redis.exceptions
 from redis.asyncio.lock import Lock
@@ -28,7 +28,11 @@ async def auto_extend_lock(lock: Lock) -> None:
         raise LockLostError(lock=lock) from exc
 
 
-async def handle_redis_returns_union_types(result: Any | Awaitable[Any]) -> Any:
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+async def handle_redis_returns_union_types(result: R | Awaitable[R]) -> R:
     """Used to handle mypy issues with redis 5.x return types"""
     if isinstance(result, Awaitable):
         return await result