rename

sanderegg · sanderegg · commit cea3e8489359 · 2025-01-09T18:27:55.000+01:00
diff --git a/packages/service-library/src/servicelib/async_utils.py b/packages/service-library/src/servicelib/async_utils.py
@@ -7,9 +7,7 @@
 from contextlib import suppress
 from dataclasses import dataclass
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Final, ParamSpec, TypeVar
-
-from tenacity import before_sleep_log, retry, stop_after_attempt
+from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
 from . import tracing
 from .utils_profiling_middleware import dont_profile, is_profiling, profile_context
@@ -234,30 +232,20 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
     return decorator
 
 
-_MAX_TASK_CANCELLATION_ATTEMPTS: Final[int] = 3
-
-
-@retry(
-    reraise=True,
-    stop=stop_after_attempt(_MAX_TASK_CANCELLATION_ATTEMPTS),
-    before_sleep=before_sleep_log(_logger, logging.WARNING, exc_info=True),
-)
-async def retried_cancel_task(
+async def cancel_wait_task(
     task: asyncio.Task,
     *,
-    timeout: float | None = None,  # noqa: ASYNC109
+    max_delay: float | None = None,
 ) -> None:
-    """Reliable task cancellation. Some libraries will just hang without
-    cancelling the task. The operation is retried _MAX_TASK_CANCELLATION_ATTEMPTS times
-    in case of timeout.
+    """Cancel a asyncio.Task and waits for it to finish.
 
     :param task: task to be canceled
-    :param timeout_delay: duration (in seconds) to wait before giving
-        up the cancellation. If None it waits forever. Will be repeated _MAX_TASK_CANCELLATION_ATTEMPTS times.
+    :param max_delay: duration (in seconds) to wait before giving
+        up the cancellation. If None it waits forever.
     :raises TimeoutError: raised if cannot cancel the task.
     """
 
     task.cancel()
-    async with asyncio.timeout(timeout):
+    async with asyncio.timeout(max_delay):
         with contextlib.suppress(asyncio.CancelledError):
             await task
diff --git a/packages/service-library/src/servicelib/background_task.py b/packages/service-library/src/servicelib/background_task.py
@@ -9,7 +9,7 @@
 from tenacity import TryAgain, before_sleep_log, retry, retry_if_exception_type
 from tenacity.wait import wait_fixed
 
-from .async_utils import retried_cancel_task, with_delay
+from .async_utils import cancel_wait_task, with_delay
 from .logging_utils import log_context
 
 _logger = logging.getLogger(__name__)
@@ -134,6 +134,4 @@ async def periodic_task(
         if asyncio_task is not None:
             # NOTE: this stopping is shielded to prevent the cancellation to propagate
             # into the stopping procedure
-            await asyncio.shield(
-                retried_cancel_task(asyncio_task, timeout=stop_timeout)
-            )
+            await asyncio.shield(cancel_wait_task(asyncio_task, max_delay=stop_timeout))
diff --git a/packages/service-library/tests/redis/test_decorators.py b/packages/service-library/tests/redis/test_decorators.py
@@ -11,7 +11,7 @@
 import arrow
 import pytest
 from faker import Faker
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.redis import (
     CouldNotAcquireLockError,
     RedisClientSDK,
@@ -217,7 +217,7 @@ async def _assert_task_completes_once(
 
         await _assert_on_sleep_done(sleep_events, stop_after=stop_after)
 
-        await retried_cancel_task(started_task, timeout=5)
+        await cancel_wait_task(started_task, max_delay=5)
 
         events_timestamps: tuple[float, ...] = tuple(
             x.args[0].timestamp() for x in sleep_events.call_args_list
diff --git a/packages/service-library/tests/test_background_task.py b/packages/service-library/tests/test_background_task.py
@@ -14,7 +14,7 @@
 import pytest
 from faker import Faker
 from pytest_mock.plugin import MockerFixture
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import (  # Assuming the module is imported correctly
     create_periodic_task,
     periodic,
@@ -73,7 +73,7 @@ async def _creator(
     yield _creator
     # cleanup
     await asyncio.gather(
-        *(retried_cancel_task(t, timeout=stop_task_timeout) for t in created_tasks)
+        *(cancel_wait_task(t, max_delay=stop_task_timeout) for t in created_tasks)
     )
 
 
diff --git a/services/agent/src/simcore_service_agent/services/volumes_manager.py b/services/agent/src/simcore_service_agent/services/volumes_manager.py
@@ -9,7 +9,7 @@
 from fastapi import FastAPI
 from models_library.projects_nodes_io import NodeID
 from pydantic import NonNegativeFloat
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.logging_utils import log_context
@@ -61,10 +61,10 @@ async def shutdown(self) -> None:
         await self.docker.close()
 
         if self._task_bookkeeping:
-            await retried_cancel_task(self._task_bookkeeping)
+            await cancel_wait_task(self._task_bookkeeping)
 
         if self._task_periodic_volume_cleanup:
-            await retried_cancel_task(self._task_periodic_volume_cleanup)
+            await cancel_wait_task(self._task_periodic_volume_cleanup)
 
     async def _bookkeeping_task(self) -> None:
         with log_context(_logger, logging.DEBUG, "volume bookkeeping"):
diff --git a/services/api-server/src/simcore_service_api_server/core/_prometheus_instrumentation.py b/services/api-server/src/simcore_service_api_server/core/_prometheus_instrumentation.py
@@ -6,7 +6,7 @@
 from fastapi import FastAPI
 from prometheus_client import CollectorRegistry, Gauge
 from pydantic import PositiveInt
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.prometheus_instrumentation import (
     setup_prometheus_instrumentation as setup_rest_instrumentation,
@@ -80,7 +80,7 @@ async def on_startup() -> None:
     async def on_shutdown() -> None:
         assert app.state.instrumentation_task  # nosec
         with log_catch(_logger, reraise=False):
-            await retried_cancel_task(app.state.instrumentation_task)
+            await cancel_wait_task(app.state.instrumentation_task)
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)
diff --git a/services/api-server/src/simcore_service_api_server/core/health_checker.py b/services/api-server/src/simcore_service_api_server/core/health_checker.py
@@ -9,7 +9,7 @@
 from models_library.rabbitmq_messages import LoggerRabbitMessage
 from models_library.users import UserID
 from pydantic import NonNegativeInt, PositiveFloat, PositiveInt
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.dependencies import get_app
 from servicelib.logging_utils import log_catch
@@ -63,8 +63,8 @@ async def setup(self, health_check_task_period_seconds: PositiveFloat):
     async def teardown(self):
         if self._background_task:
             with log_catch(_logger, reraise=False):
-                await retried_cancel_task(
-                    self._background_task, timeout=self._timeout_seconds
+                await cancel_wait_task(
+                    self._background_task, max_delay=self._timeout_seconds
                 )
         await self._log_distributor.deregister(job_id=self._dummy_job_id)
 
diff --git a/services/autoscaling/src/simcore_service_autoscaling/modules/auto_scaling_task.py b/services/autoscaling/src/simcore_service_autoscaling/modules/auto_scaling_task.py
@@ -3,7 +3,7 @@
 from typing import Final
 
 from fastapi import FastAPI
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
@@ -44,7 +44,7 @@ async def _startup() -> None:
 
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
-        await retried_cancel_task(app.state.autoscaler_task)
+        await cancel_wait_task(app.state.autoscaler_task)
 
     return _stop
 
diff --git a/services/autoscaling/src/simcore_service_autoscaling/modules/buffer_machines_pool_task.py b/services/autoscaling/src/simcore_service_autoscaling/modules/buffer_machines_pool_task.py
@@ -3,7 +3,7 @@
 from typing import Final
 
 from fastapi import FastAPI
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
@@ -44,7 +44,7 @@ async def _startup() -> None:
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
         if hasattr(app.state, "buffers_pool_task"):
-            await retried_cancel_task(app.state.buffers_pool_task)
+            await cancel_wait_task(app.state.buffers_pool_task)
 
     return _stop
 
diff --git a/services/clusters-keeper/src/simcore_service_clusters_keeper/modules/clusters_management_task.py b/services/clusters-keeper/src/simcore_service_clusters_keeper/modules/clusters_management_task.py
@@ -3,7 +3,7 @@
 from collections.abc import Awaitable, Callable
 
 from fastapi import FastAPI
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.redis import exclusive
 
@@ -37,7 +37,7 @@ async def _startup() -> None:
 
 def on_app_shutdown(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _stop() -> None:
-        await retried_cancel_task(app.state.clusters_cleaning_task, timeout=5)
+        await cancel_wait_task(app.state.clusters_cleaning_task, max_delay=5)
 
     return _stop
 
diff --git a/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_manager.py b/services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_manager.py
@@ -6,7 +6,7 @@
 from fastapi import FastAPI
 from models_library.projects import ProjectID
 from models_library.users import UserID
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.exception_utils import silence_exceptions
 from servicelib.logging_utils import log_context
@@ -165,4 +165,4 @@ async def setup_manager(app: FastAPI) -> None:
 
 
 async def shutdown_manager(app: FastAPI) -> None:
-    await retried_cancel_task(app.state.scheduler_manager)
+    await cancel_wait_task(app.state.scheduler_manager)
diff --git a/services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/_core/_scheduler.py b/services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/_core/_scheduler.py
@@ -41,7 +41,7 @@
 from models_library.users import UserID
 from models_library.wallets import WalletID
 from pydantic import NonNegativeFloat
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.fastapi.long_running_tasks.client import ProgressCallback
 from servicelib.fastapi.long_running_tasks.server import TaskProgress
@@ -126,7 +126,7 @@ async def shutdown(self) -> None:
         self._to_observe = {}
 
         if self._scheduler_task is not None:
-            await retried_cancel_task(self._scheduler_task, timeout=5)
+            await cancel_wait_task(self._scheduler_task, max_delay=5)
             self._scheduler_task = None
 
         if self._trigger_observation_queue_task is not None:
@@ -365,7 +365,7 @@ async def mark_service_for_removal(
                     self._service_observation_task[service_name]
                 )
                 if isinstance(service_task, asyncio.Task):
-                    await retried_cancel_task(service_task, timeout=10)
+                    await cancel_wait_task(service_task, max_delay=10)
 
             if skip_observation_recreation:
                 return
diff --git a/services/director-v2/src/simcore_service_director_v2/utils/base_distributed_identifier.py b/services/director-v2/src/simcore_service_director_v2/utils/base_distributed_identifier.py
@@ -5,7 +5,7 @@
 from typing import Final, Generic, TypeVar
 
 from pydantic import NonNegativeInt
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.redis import RedisClientSDK
@@ -76,7 +76,7 @@ async def setup(self) -> None:
 
     async def shutdown(self) -> None:
         if self._cleanup_task:
-            await retried_cancel_task(self._cleanup_task, timeout=5)
+            await cancel_wait_task(self._cleanup_task, max_delay=5)
 
     @classmethod
     def class_path(cls) -> str:
diff --git a/services/director/src/simcore_service_director/registry_proxy.py b/services/director/src/simcore_service_director/registry_proxy.py
@@ -10,7 +10,7 @@
 import httpx
 from aiocache import Cache, SimpleMemoryCache  # type: ignore[import-untyped]
 from fastapi import FastAPI, status
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.utils import limited_as_completed
@@ -257,7 +257,7 @@ async def on_startup() -> None:
 
     async def on_shutdown() -> None:
         if app.state.auto_cache_task:
-            await retried_cancel_task(app.state.auto_cache_task)
+            await cancel_wait_task(app.state.auto_cache_task)
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)
diff --git a/services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_monitor.py b/services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_monitor.py
@@ -7,7 +7,7 @@
 from fastapi import FastAPI
 from models_library.projects_nodes_io import NodeID
 from pydantic import NonNegativeFloat, NonNegativeInt
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.redis import start_exclusive_periodic_task
 from servicelib.utils import limited_gather
 from settings_library.redis import RedisDatabase
@@ -143,4 +143,4 @@ async def setup(self) -> None:
 
     async def shutdown(self) -> None:
         if getattr(self.app.state, "status_monitor_background_task", None):
-            await retried_cancel_task(self.app.state.status_monitor_background_task)
+            await cancel_wait_task(self.app.state.status_monitor_background_task)
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/outputs/_manager.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/outputs/_manager.py
@@ -12,7 +12,7 @@
 from models_library.rabbitmq_messages import ProgressType
 from pydantic import PositiveFloat
 from servicelib import progress_bar
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.logging_utils import log_catch, log_context
 from simcore_sdk.node_ports_common.file_io_utils import LogRedirectCB
@@ -205,8 +205,8 @@ async def shutdown(self) -> None:
         with log_context(_logger, logging.INFO, f"{OutputsManager.__name__} shutdown"):
             await self._uploading_task_cancel()
             if self._task_scheduler_worker is not None:
-                await retried_cancel_task(
-                    self._task_scheduler_worker, timeout=self.task_monitor_interval_s
+                await cancel_wait_task(
+                    self._task_scheduler_worker, max_delay=self.task_monitor_interval_s
                 )
 
     async def port_key_content_changed(self, port_key: str) -> None:
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/prometheus_metrics.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/prometheus_metrics.py
@@ -9,7 +9,7 @@
 from fastapi import FastAPI, status
 from models_library.callbacks_mapping import CallbacksMapping, UserServiceCommand
 from pydantic import BaseModel, NonNegativeFloat, NonNegativeInt
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.logging_utils import log_context
 from servicelib.sequences_utils import pairwise
 from simcore_service_dynamic_sidecar.core.errors import (
@@ -143,8 +143,8 @@ async def start(self) -> None:
     async def stop(self) -> None:
         with log_context(_logger, logging.INFO, "shutdown service metrics recovery"):
             if self._metrics_recovery_task:
-                await retried_cancel_task(
-                    self._metrics_recovery_task, timeout=_TASK_CANCELLATION_TIMEOUT_S
+                await cancel_wait_task(
+                    self._metrics_recovery_task, max_delay=_TASK_CANCELLATION_TIMEOUT_S
                 )
 
 
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/resource_tracking/_core.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/resource_tracking/_core.py
@@ -14,7 +14,7 @@
 from models_library.services import ServiceType
 from models_library.services_creation import CreateServiceMetricsAdditionalParams
 from pydantic import NonNegativeFloat
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.logging_utils import log_context
 
@@ -62,8 +62,8 @@ async def _start_heart_beat_task(app: FastAPI) -> None:
 async def stop_heart_beat_task(app: FastAPI) -> None:
     resource_tracking: ResourceTrackingState = app.state.resource_tracking
     if resource_tracking.heart_beat_task:
-        await retried_cancel_task(
-            resource_tracking.heart_beat_task, timeout=_STOP_WORKER_TIMEOUT_S
+        await cancel_wait_task(
+            resource_tracking.heart_beat_task, max_delay=_STOP_WORKER_TIMEOUT_S
         )
 
 
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/system_monitor/_disk_usage.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/system_monitor/_disk_usage.py
@@ -14,7 +14,7 @@
 )
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
-from servicelib.async_utils import retried_cancel_task
+from servicelib.async_utils import cancel_wait_task
 from servicelib.background_task import create_periodic_task
 from servicelib.logging_utils import log_context
 from servicelib.utils import logged_gather
@@ -181,7 +181,7 @@ async def setup(self) -> None:
 
     async def shutdown(self) -> None:
         if self._monitor_task:
-            await retried_cancel_task(self._monitor_task)
+            await cancel_wait_task(self._monitor_task)
 
     def set_disk_usage_for_path(self, overwrite_usage: dict[str, DiskUsage]) -> None:
         """
diff --git a/services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks_setup.py b/services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks_setup.py
diff --git a/services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/background_task_periodic_heartbeat_check_setup.py b/services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/background_task_periodic_heartbeat_check_setup.py
diff --git a/services/storage/src/simcore_service_storage/dsm_cleaner.py b/services/storage/src/simcore_service_storage/dsm_cleaner.py
