Cleanup

Signed-off-by: Sergio Herrera <627709+seherv@users.noreply.github.com>

Author: seherv

ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/worker.py

@@ -352,6 +352,7 @@ def __init__(
             self._interceptors = None
 
         self._async_worker_manager = _AsyncWorkerManager(self._concurrency_options, self._logger)
+        self._activity_executor = _ActivityExecutor(self._logger)
 
     @property
     def concurrency_options(self) -> ConcurrencyOptions:
@@ -1084,8 +1085,7 @@ def _execute_activity(
         instance_id = req.workflowInstance.instanceId
         with self._activity_span(req, instance_id):
             try:
-                executor = _ActivityExecutor(self._logger)
-                result = executor.execute(
+                result = self._activity_executor.execute(
                     fn,
                     instance_id,
                     req.name,
@@ -1114,8 +1114,7 @@ async def _execute_activity_async(
         instance_id = req.workflowInstance.instanceId
         with self._activity_span(req, instance_id):
             try:
-                executor = _ActivityExecutor(self._logger)
-                result = await executor.execute_async(
+                result = await self._activity_executor.execute_async(
                     fn,
                     instance_id,
                     req.name,
@@ -1143,10 +1142,11 @@ async def _execute_activity_async(
                     instance_id,
                 )
             except RuntimeError as exc:
-                # Only swallow if the manager thread pool was shut down (worker is tearing
-                # down). The sidecar will re-dispatch the work item once the worker reconnects.
-                # Other RuntimeErrors are unexpected and bubble up to the work-item processor.
-                if not self._async_worker_manager._shutdown:
+                # Swallow only when the thread pool itself is shut down (worker tearing down).
+                # Other RuntimeErrors are unexpected and propagate to the work-item processor.
+                # The sidecar will re-dispatch this work item once the worker reconnects.
+                pool = self._async_worker_manager.thread_pool
+                if not getattr(pool, '_shutdown', False):
                     raise
                 self._logger.warning(
                     f"Could not deliver activity response for '{req.name}#{req.taskId}': "

ext/dapr-ext-workflow/dapr/ext/workflow/workflow_runtime.py

@@ -68,32 +68,34 @@ def _make_activity_wrapper(fn: Activity, logger: Logger) -> ActivityWrapper:
     """
     accepts_input, input_model = _model_protocol.resolve_input(fn)
 
+    def _call_args(ctx: task.ActivityContext, inp: object | None) -> tuple:
+        wf_ctx = WorkflowActivityContext(ctx)
+        if not accepts_input:
+            return (wf_ctx,)
+        return (wf_ctx, _coerce_activity_input(inp, input_model))
+
+    def _log_failure(ctx: task.ActivityContext, exc: Exception) -> None:
+        activity_id = getattr(ctx, 'task_id', 'unknown')
+        logger.warning(f'Activity execution failed - task_id: {activity_id}, error: {exc}')
+
     if _is_async_callable(fn):
 
         async def async_activity_wrapper(
             ctx: task.ActivityContext, inp: object | None = None
         ) -> object:
-            activity_id = getattr(ctx, 'task_id', 'unknown')
             try:
-                wf_ctx = WorkflowActivityContext(ctx)
-                if not accepts_input:
-                    return await fn(wf_ctx)
-                return await fn(wf_ctx, _coerce_activity_input(inp, input_model))
-            except Exception as e:
-                logger.warning(f'Activity execution failed - task_id: {activity_id}, error: {e}')
+                return await fn(*_call_args(ctx, inp))
+            except Exception as exc:
+                _log_failure(ctx, exc)
                 raise
 
         return async_activity_wrapper
 
     def sync_activity_wrapper(ctx: task.ActivityContext, inp: object | None = None) -> object:
-        activity_id = getattr(ctx, 'task_id', 'unknown')
         try:
-            wf_ctx = WorkflowActivityContext(ctx)
-            if not accepts_input:
-                return fn(wf_ctx)
-            return fn(wf_ctx, _coerce_activity_input(inp, input_model))
-        except Exception as e:
-            logger.warning(f'Activity execution failed - task_id: {activity_id}, error: {e}')
+            return fn(*_call_args(ctx, inp))
+        except Exception as exc:
+            _log_failure(ctx, exc)
             raise
 
     return sync_activity_wrapper