Merge branch 'main' into remove-mechanical-markdown

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

Author: seherv

.github/scripts/automerge.py

@@ -12,6 +12,7 @@
 #
 
 name: Backport
+
 on:
   pull_request_target:
     types:
@@ -20,18 +21,6 @@ on:
 
 jobs:
   backport:
-    name: Backport
-    runs-on: ubuntu-latest
-    if: >
-      github.event.pull_request.merged
-      && (
-        github.event.action == 'closed'
-        || (
-          github.event.action == 'labeled'
-          && contains(github.event.label.name, 'backport')
-        )
-      )
-    steps:
-      - uses: tibdex/backport@9565281eda0731b1d20c4025c43339fb0a23812e
-        with:
-          github_token: ${{ secrets.DAPR_BOT_TOKEN }}
+    uses: dapr/.github/.github/workflows/backport.yaml@main
+    secrets:
+      dapr_bot_token: ${{ secrets.DAPR_BOT_TOKEN }}

.github/workflows/backport.yaml

@@ -17,53 +17,21 @@ on:
   schedule:
     - cron: '*/10 * * * *'
   workflow_dispatch:
+
 jobs:
   automerge:
-    if: github.repository_owner == 'dapr'
-    name: Automerge and update PRs.
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout repo
-        uses: actions/checkout@v6
-      - name: Install dependencies
-        run: pip install PyGithub
-      - name: Automerge and update
-        env:
-          MAINTAINERS: berndverst,wcs1only
-          GITHUB_TOKEN: ${{ secrets.DAPR_BOT_TOKEN }}
-        run: python ./.github/scripts/automerge.py
+    uses: dapr/.github/.github/workflows/automerge.yaml@main
+    with:
+      maintainer-teams: python-sdk-maintainers
+    secrets:
+      dapr_bot_token: ${{ secrets.DAPR_BOT_TOKEN }}
+
   prune_stale:
-    name: Prune Stale
-    runs-on: ubuntu-latest
-    steps:
-    - name: Prune Stale
-      uses: actions/stale@v5
-      with:
-        repo-token: ${{ secrets.DAPR_BOT_TOKEN }}
-        # Different amounts of days for issues/PRs are not currently supported but there is a PR
-        # open for it: https://github.com/actions/stale/issues/214
-        days-before-stale: 60
-        days-before-close: 7
-        stale-issue-message: >
-          This issue has been automatically marked as stale because it has not had activity in the
-          last 60 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue, help wanted or triaged/resolved) or other activity
-          occurs. Thank you for your contributions.
-        close-issue-message: >
-          This issue has been automatically closed because it has not had activity in the
-          last 67 days. If this issue is still valid, please ping a maintainer and ask them to label it as pinned, good first issue, help wanted or triaged/resolved.
-          Thank you for your contributions.
-        stale-pr-message: >
-          This pull request has been automatically marked as stale because it has not had
-          activity in the last 60 days. It will be closed in 7 days if no further activity occurs. Please
-          feel free to give a status update now, ping for review, or re-open when it's ready.
-          Thank you for your contributions!
-        close-pr-message: >
-          This pull request has been automatically closed because it has not had
-          activity in the last 67 days. Please feel free to give a status update now, ping for review, or re-open when it's ready.
-          Thank you for your contributions!
-        stale-issue-label: 'stale'
-        exempt-issue-labels: 'pinned,good first issue,help wanted,triaged/resolved'
-        stale-pr-label: 'stale'
-        exempt-pr-labels: 'pinned'
-        operations-per-run: 500
-        ascending: true
+    uses: dapr/.github/.github/workflows/prune-stale.yaml@main
+    with:
+      days-before-issue-stale: 60
+      days-before-pr-stale: 60
+      days-before-issue-close: 7
+      days-before-pr-close: 7
+    secrets:
+      dapr_bot_token: ${{ secrets.DAPR_BOT_TOKEN }}

.github/workflows/dapr-bot-schedule.yml

@@ -331,6 +331,7 @@ def __init__(
         self._current_channel: Optional[grpc.Channel] = None  # Store channel reference for cleanup
         self._channel_cleanup_threads: list[threading.Thread] = []  # Deferred channel close threads
         self._stream_ready = threading.Event()
+        self._runLoop: Optional[Thread] = None
         # Use provided concurrency options or create default ones
         self._concurrency_options = (
             concurrency_options if concurrency_options is not None else ConcurrencyOptions()
@@ -387,8 +388,13 @@ def run_loop():
         self._logger.info(f'Starting gRPC worker that connects to {self._host_address}')
         self._runLoop = Thread(target=run_loop, name='WorkerRunLoop')
         self._runLoop.start()
-        if not self._stream_ready.wait(timeout=10):
-            raise RuntimeError('Failed to establish work item stream connection within 10 seconds')
+        while not self._stream_ready.wait(timeout=1):
+            if self._shutdown.is_set():
+                raise RuntimeError('Worker was stopped before the work item stream was established')
+            if not self._runLoop.is_alive():
+                raise RuntimeError(
+                    'Worker run loop exited before the work item stream was established'
+                )
         self._is_running = True
 
     async def _keepalive_loop(self, stub):
@@ -801,7 +807,9 @@ def _deferred_close():
 
     def stop(self):
         """Stops the worker and waits for any pending work items to complete."""
-        if not self._is_running:
+        # Guards on _runLoop rather than _is_running so stop() can unblock a start()
+        # that is still waiting for the work item stream to be established.
+        if self._runLoop is None:
             return
 
         self._logger.info('Stopping gRPC worker...')
@@ -833,6 +841,7 @@ def stop(self):
         self._async_worker_manager.shutdown()
         self._logger.info('Worker shutdown completed')
         self._is_running = False
+        self._runLoop = None
 
     # TODO: This should be removed in the future as we do handle grpc errs
     def _handle_grpc_execution_error(self, rpc_error: grpc.RpcError, request_type: str):

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

@@ -1,6 +1,10 @@
+import asyncio
+import threading
+import time
 from unittest.mock import MagicMock, patch
 
 import grpc
+import pytest
 from dapr.ext.workflow._durabletask.worker import TaskHubGrpcWorker
 
 
@@ -146,3 +150,65 @@ def test_deferred_close_prunes_finished_threads():
     worker._channel_cleanup_threads[-1].join(timeout=2)
     # Only the still-alive (or just-finished ch2) thread remains; ch1's was pruned
     assert len(worker._channel_cleanup_threads) <= 1
+
+
+def test_stop_before_start_is_noop():
+    """stop() is safe to call before start() — _runLoop is None, no AttributeError."""
+    worker = TaskHubGrpcWorker()
+    with patch.object(worker._shutdown, 'set') as shutdown_set:
+        worker.stop()
+        shutdown_set.assert_not_called()
+
+
+def test_stop_is_idempotent():
+    """A second stop() returns early because _runLoop was cleared by the first."""
+    worker = _make_running_worker()
+    worker._current_channel = MagicMock()
+    worker.stop()
+    assert worker._runLoop is None
+    with patch.object(worker._shutdown, 'set') as shutdown_set:
+        worker.stop()
+        shutdown_set.assert_not_called()
+
+
+def test_start_raises_when_run_loop_exits_early():
+    """start() raises RuntimeError if the run loop thread exits before _stream_ready is set."""
+    worker = TaskHubGrpcWorker()
+
+    async def fast_exit():
+        return
+
+    with patch.object(worker, '_async_run_loop', side_effect=fast_exit):
+        with pytest.raises(RuntimeError, match='Worker run loop exited'):
+            worker.start()
+
+
+def test_start_raises_when_stopped_during_startup():
+    """stop() unblocks a start() that is waiting for _stream_ready; start() raises."""
+    worker = TaskHubGrpcWorker()
+
+    async def wait_for_shutdown():
+        # Block without setting _stream_ready so start() stays in its wait loop.
+        while not worker._shutdown.is_set():
+            await asyncio.sleep(0.05)
+
+    errors = []
+
+    def _start():
+        try:
+            worker.start()
+        except Exception as e:  # noqa: BLE001
+            errors.append(e)
+
+    with patch.object(worker, '_async_run_loop', side_effect=wait_for_shutdown):
+        t = threading.Thread(target=_start)
+        t.start()
+        # Let start() enter its wait loop (timeout=1 per iteration).
+        time.sleep(1.2)
+        worker.stop()
+        t.join(timeout=5)
+
+    assert not t.is_alive(), 'start() did not return after stop()'
+    assert len(errors) == 1, f'Expected exactly one error, got: {errors}'
+    assert isinstance(errors[0], RuntimeError)
+    assert 'Worker was stopped' in str(errors[0])