Fix flaky sync client recreate tests by joining the recreate thread (#214)

The synchronous client's fire-and-forget recreate thread signals
`_recreate_done_event` from inside `_run_recreate`'s `finally` block, so
the event becomes set while the thread is still alive.
`_schedule_recreate` single-flights on `existing.is_alive()` and its early
return sits *before* `_recreate_done_event.clear()`. A trigger issued in
that window is therefore dropped and the event is left set from the
previous recreate, so the next `wait()` returns immediately off the stale
signal and assertions about the following recreate fail.

Two tests chained recreates through this pattern and were intermittently
flaky under CPU contention:

- test_sync_client_close_closes_all_retired_sdk_channels_immediately
- test_sync_client_recreate_cooldown_prevents_immediate_repeated_recreation

Add an `await_sync_recreate` helper that waits on the event and then joins
the recreate thread, so `is_alive()` is deterministically False before the
next trigger is issued, and use it at all five wait sites.

This is a test-synchronization defect only; the production single-flight
behaviour is correct and intentional, so no production code changes. The
async client is unaffected -- it gates on `asyncio.Task.done()`.

Fixes #212

Co-authored-by: Bernd Verst <beverst@microsoft.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 677577fc-869c-463f-9243-0f4802a9f040

Author: 3 people

tests/durabletask/test_client.py

@@ -214,9 +214,10 @@ def install_resilient_test_stubs(client):
     call so newly created stubs continue to participate in failure tracking.
 
     The interceptor's ``_on_recreate`` callback is intentionally left alone:
-    it is the client's ``_schedule_recreate`` (fire-and-forget), and tests
-    that need to observe the recreate's completion can wait on
-    ``client._recreate_done_event``.
+    it is the client's ``_schedule_recreate`` (fire-and-forget). Synchronous
+    tests that chain recreates must observe completion via
+    ``await_sync_recreate``; waiting on ``client._recreate_done_event`` alone
+    is not sufficient (see that helper's docstring).
     """
     is_async = inspect.iscoroutinefunction(client._maybe_recreate_channel)
     wrapper_cls = _ResilientAsyncTestStub if is_async else _ResilientSyncTestStub
@@ -241,6 +242,28 @@ def wrapped_recreate():
     client._maybe_recreate_channel = wrapped_recreate
 
 
+def await_sync_recreate(client: TaskHubGrpcClient, timeout: float = 5.0) -> None:
+    """Block until the synchronous client's recreate thread has fully exited.
+
+    ``_recreate_done_event`` is set by ``_run_recreate`` from *inside* the
+    recreate thread, so the event is signalled while that thread is still
+    alive. ``_schedule_recreate`` single-flights on ``is_alive()``, so a
+    trigger issued in that window is dropped *before* it reaches
+    ``_recreate_done_event.clear()`` -- leaving the event set from the
+    previous recreate. A later ``wait()`` would then return immediately off
+    that stale signal, and assertions about the next recreate would fail.
+
+    Tests that chain recreates must therefore wait on the same condition the
+    single-flight guard reads. Joining the thread makes ``is_alive()``
+    deterministically ``False`` before the next trigger is issued.
+    """
+    assert client._recreate_done_event.wait(timeout=timeout)
+    recreate_thread = client._recreate_thread
+    if recreate_thread is not None:
+        recreate_thread.join(timeout=timeout)
+        assert not recreate_thread.is_alive()
+
+
 class FakePayloadStore(PayloadStore):
     TOKEN_PREFIX = 'fake://'
 
@@ -669,10 +692,10 @@ def test_sync_client_close_closes_all_retired_sdk_channels_immediately():
         # Wait for the first fire-and-forget recreate to complete so the
         # single-flight guard in _schedule_recreate does not drop the second
         # trigger.
-        assert client._recreate_done_event.wait(timeout=5.0)
+        await_sync_recreate(client)
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
-        assert client._recreate_done_event.wait(timeout=5.0)
+        await_sync_recreate(client)
 
         client.close()
 
@@ -912,21 +935,21 @@ def test_sync_client_recreate_cooldown_prevents_immediate_repeated_recreation():
             client.get_orchestration_state("abc")
         # Wait for the fire-and-forget recreate to complete before asserting
         # the channel was swapped.
-        assert client._recreate_done_event.wait(timeout=5.0)
+        await_sync_recreate(client)
         assert client._channel is second_channel
         assert mock_get_channel.call_count == 2
 
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
         # Cooldown should fire-and-forget but exit without recreating; wait
         # for the no-op recreate to complete so the assertion is deterministic.
-        assert client._recreate_done_event.wait(timeout=5.0)
+        await_sync_recreate(client)
         assert client._channel is second_channel
         assert mock_get_channel.call_count == 2
 
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
-        assert client._recreate_done_event.wait(timeout=5.0)
+        await_sync_recreate(client)
         assert client._channel is third_channel
 
     expected_channel_call = call(