Revert debug stuff. Update core to latest

Author: tconley1428

temporalio/bridge/Cargo.lock

@@ -27,14 +27,14 @@ pyo3 = { version = "0.25", features = [
 ] }
 pyo3-async-runtimes = { version = "0.25", features = ["tokio-runtime"] }
 pythonize = "0.25"
-temporal-client = { git = "https://github.com/temporalio/sdk-core", branch = "shutdown_notify" }
-temporal-sdk-core = { git = "https://github.com/temporalio/sdk-core", branch = "shutdown_notify", features = [
+temporal-client = { version = "0.1.0", path = "./sdk-core/client" }
+temporal-sdk-core = { version = "0.1.0", path = "./sdk-core/core", features = [
     "ephemeral-server",
 ] }
-temporal-sdk-core-api = { git = "https://github.com/temporalio/sdk-core", branch = "shutdown_notify",  features = [
+temporal-sdk-core-api = { version = "0.1.0", path = "./sdk-core/core-api", features = [
     "envconfig",
 ] }
-temporal-sdk-core-protos = { git = "https://github.com/temporalio/sdk-core", branch = "shutdown_notify" }
+temporal-sdk-core-protos = { version = "0.1.0", path = "./sdk-core/sdk-core-protos" }
 tokio = "1.26"
 tokio-stream = "0.1"
 tonic = "0.13"

temporalio/bridge/Cargo.toml

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import logging
 from dataclasses import dataclass
 from typing import (
     TYPE_CHECKING,
@@ -43,8 +42,6 @@
 )
 from temporalio.bridge.temporal_sdk_bridge import PollShutdownError  # type: ignore
 
-logger = logging.getLogger(__name__)
-
 
 @dataclass
 class WorkerConfig:
@@ -222,8 +219,6 @@ async def poll_activity_task(
         self,
     ) -> temporalio.bridge.proto.activity_task.ActivityTask:
         """Poll for an activity task."""
-        logger.info("Polling for activity task")
-
         return temporalio.bridge.proto.activity_task.ActivityTask.FromString(
             await self._ref.poll_activity_task()
         )
@@ -247,7 +242,6 @@ async def complete_activity_task(
         self, comp: temporalio.bridge.proto.ActivityTaskCompletion
     ) -> None:
         """Complete an activity task."""
-        logger.info("Completing activity task: %s", comp.task_token)
         await self._ref.complete_activity_task(comp.SerializeToString())
 
     async def complete_nexus_task(

temporalio/bridge/sdk-core

@@ -169,13 +169,10 @@ async def raise_from_exception_queue() -> NoReturn:
                 else:
                     raise RuntimeError(f"Unrecognized activity task: {task}")
             except temporalio.bridge.worker.PollShutdownError:  # type: ignore[reportPrivateLocalImportUsage]
-                logger.info("Activity worker finished1")
                 exception_task.cancel()
-                logger.info("Activity worker finished2")
                 return
             except Exception as err:
                 exception_task.cancel()
-                logger.info("Activity worker failed: %s", err)
                 raise RuntimeError("Activity worker failed") from err
 
     def notify_shutdown(self) -> None:
@@ -186,18 +183,14 @@ def notify_shutdown(self) -> None:
     async def drain_poll_queue(self) -> None:
         while True:
             try:
-                logger.info("Poll activity tasks")
-
                 # Just take all tasks and say we can't handle them
                 task = await self._bridge_worker().poll_activity_task()
                 completion = temporalio.bridge.proto.ActivityTaskCompletion(  # type: ignore[reportAttributeAccessIssue]
                     task_token=task.task_token
                 )
                 completion.result.failed.failure.message = "Worker shutting down"
-                logger.info("Complete activity task")
                 await self._bridge_worker().complete_activity_task(completion)
             except temporalio.bridge.worker.PollShutdownError:  # type: ignore[reportPrivateLocalImportUsage]
-                logger.info("Return from drain")
                 return
 
     # Only call this after run()/drain_poll_queue() have returned. This will not

temporalio/bridge/worker.py

@@ -755,34 +755,25 @@ async def raise_on_shutdown():
         # Initiate core worker shutdown
         self._bridge_worker.initiate_shutdown()
 
-        logger.info("Bridge shut down")
-
         # If any worker task had an exception, replace that task with a queue drain
         for worker, task in tasks.items():
             if worker and task.done() and task.exception():
-                logger.info("Adding a drain task")
                 tasks[worker] = asyncio.create_task(worker.drain_poll_queue())
 
-        logger.info("Task drain done")
-
         # Notify shutdown occurring
         if self._activity_worker:
             self._activity_worker.notify_shutdown()
         if self._workflow_worker:
             self._workflow_worker.notify_shutdown()
-        logger.info("workers notified")
 
         # Wait for all tasks to complete (i.e. for poller loops to stop)
         await asyncio.wait(tasks.values())
-        logger.info("tasks waited")
-
         # Sometimes both workers throw an exception and since we only take the
         # first, Python may complain with "Task exception was never retrieved"
         # if we don't get the others. Therefore we call cancel on each task
         # which suppresses this.
         for task in tasks.values():
             task.cancel()
-        logger.info("tasks cancelled")
 
         # Let all activity / nexus operations completions finish. We cannot guarantee that
         # because poll shutdown completed (which means activities/operations completed)
@@ -791,7 +782,6 @@ async def raise_on_shutdown():
             await self._activity_worker.wait_all_completed()
         if self._nexus_worker:
             await self._nexus_worker.wait_all_completed()
-        logger.info("Waited for workers")
 
         # Do final shutdown
         try:
@@ -800,7 +790,6 @@ async def raise_on_shutdown():
             # Ignore errors here that can arise in some tests where the bridge
             # worker still has a reference
             pass
-        logger.info("Set shutdown complete")
 
         # Mark as shutdown complete and re-raise exception if present
         self._shutdown_complete_event.set()
@@ -858,8 +847,6 @@ async def __aexit__(self, exc_type: Optional[Type[BaseException]], *args) -> Non
         if not self._async_context_run_task:
             raise RuntimeError("Never started")
         await self.shutdown()
-        logger.info("Shutdown complete")
-
         # Cancel our run task
         self._async_context_run_task.cancel()
         # Only re-raise our exception if present and exc_type is cancel

temporalio/worker/_activity.py

@@ -152,8 +152,6 @@
 with workflow.unsafe.imports_passed_through():
     import pytest
 
-logger = logging.getLogger(__name__)
-
 
 @workflow.defn
 class HelloWorkflow:
@@ -890,9 +888,7 @@ async def wait_cancel(self) -> str:
         except asyncio.CancelledError:
             return "Got cancelled error, cancelled? " + str(activity.is_cancelled())
         finally:
-            logger.info("ActivityWaitCancelNotify setting event")
             self.wait_cancel_complete.set()
-            logger.info("ActivityWaitCancelNotify set event")
 
 
 @dataclass
@@ -953,8 +949,6 @@ async def test_workflow_cancel_activity(client: Client, local: bool):
     async with new_worker(
         client, CancelActivityWorkflow, activities=[activity_inst.wait_cancel]
     ) as worker:
-        logger.info("Start first workflow")
-
         # Try cancel - confirm error and activity was sent the cancel
         handle = await client.start_workflow(
             CancelActivityWorkflow.run,
@@ -970,18 +964,12 @@ async def test_workflow_cancel_activity(client: Client, local: bool):
         async def activity_result() -> str:
             return await handle.query(CancelActivityWorkflow.activity_result)
 
-        logger.info("Wait for cancelled")
         await assert_eq_eventually(
             "Error: CancelledError", activity_result, timeout=assert_timeout
         )
-
-        logger.info("Wait for activity complete")
         await activity_inst.wait_cancel_complete.wait()
-
-        logger.info("Wait for workflow cancel")
         await handle.cancel()
 
-        logger.info("Start second workflow")
         # Wait cancel - confirm no error due to graceful cancel handling
         handle = await client.start_workflow(
             CancelActivityWorkflow.run,
@@ -993,18 +981,14 @@ async def activity_result() -> str:
             task_queue=worker.task_queue,
             task_timeout=task_timeout,
         )
-        logger.info("Wait for cancelled")
         await assert_eq_eventually(
             "Got cancelled error, cancelled? True",
             activity_result,
             timeout=assert_timeout,
         )
-        logger.info("Wait for activity complete")
         await activity_inst.wait_cancel_complete.wait()
-        logger.info("Wait for workflow cancel")
         await handle.cancel()
 
-        logger.info("Start third workflow")
         # Abandon - confirm error and that activity stays running
         handle = await client.start_workflow(
             CancelActivityWorkflow.run,
@@ -1016,16 +1000,12 @@ async def activity_result() -> str:
             task_queue=worker.task_queue,
             task_timeout=task_timeout,
         )
-        logger.info("Wait for cancelled")
         await assert_eq_eventually(
             "Error: CancelledError", activity_result, timeout=assert_timeout
         )
-        logger.info("sleep")
         await asyncio.sleep(0.5)
         assert not activity_inst.wait_cancel_complete.is_set()
-        logger.info("Wait for workflow cancel")
         await handle.cancel()
-        logger.info("Wait for activity complete")
         await activity_inst.wait_cancel_complete.wait()
 
 
@@ -2966,12 +2946,21 @@ async def waiting_signal() -> bool:
             task_queue=task_queue,
         )
 
+        # Need to wait until it has gotten halfway through, otherwise the post_patch workflow may never complete
+        async def waiting_signal() -> bool:
+            return await post_patch_handle.query(
+                PatchMemoizedWorkflowPatched.waiting_signal
+            )
+
+        await assert_eq_eventually(True, waiting_signal)
+
         # Send signal to both and check results
         await pre_patch_handle.signal(PatchMemoizedWorkflowUnpatched.signal)
         await post_patch_handle.signal(PatchMemoizedWorkflowPatched.signal)
 
         # Confirm expected values
         assert ["some-value"] == await pre_patch_handle.result()
+
         assert [
             "pre-patch",
             "some-value",
@@ -6111,22 +6100,21 @@ def __init__(
         self.main_workflow_returns_before_signal_completions = (
             main_workflow_returns_before_signal_completions
         )
-        self.ping_pong_val = 1
-        self.ping_pong_counter = 0
-        self.ping_pong_max_count = 4
+        self.run_finished = False
 
     @workflow.run
     async def run(self) -> str:
         await workflow.wait_condition(
             lambda: self.seen_first_signal and self.seen_second_signal
         )
+        self.run_finished = True
         return "workflow-result"
 
     @workflow.signal
     async def this_signal_executes_first(self):
         self.seen_first_signal = True
         if self.main_workflow_returns_before_signal_completions:
-            await self.ping_pong(lambda: self.ping_pong_val > 0)
+            await workflow.wait_condition(lambda: self.run_finished)
         raise ApplicationError(
             "Client should see this error unless doing ping-pong "
             "(in which case main coroutine returns first)"
@@ -6137,18 +6125,12 @@ async def this_signal_executes_second(self):
         await workflow.wait_condition(lambda: self.seen_first_signal)
         self.seen_second_signal = True
         if self.main_workflow_returns_before_signal_completions:
-            await self.ping_pong(lambda: self.ping_pong_val < 0)
+            await workflow.wait_condition(lambda: self.run_finished)
         raise ApplicationError("Client should never see this error!")
 
-    async def ping_pong(self, cond: Callable[[], bool]):
-        while self.ping_pong_counter < self.ping_pong_max_count:
-            await workflow.wait_condition(cond)
-            self.ping_pong_val = -self.ping_pong_val
-            self.ping_pong_counter += 1
-
 
 @workflow.defn
-class FirstCompletionCommandIsHonoredPingPongWorkflow(
+class FirstCompletionCommandIsHonoredSignalWaitWorkflow(
     FirstCompletionCommandIsHonoredWorkflow
 ):
     def __init__(self) -> None:
@@ -6177,10 +6159,10 @@ async def _do_first_completion_command_is_honored_test(
     client: Client, main_workflow_returns_before_signal_completions: bool
 ):
     workflow_cls: Union[
-        Type[FirstCompletionCommandIsHonoredPingPongWorkflow],
+        Type[FirstCompletionCommandIsHonoredSignalWaitWorkflow],
         Type[FirstCompletionCommandIsHonoredWorkflow],
     ] = (
-        FirstCompletionCommandIsHonoredPingPongWorkflow
+        FirstCompletionCommandIsHonoredSignalWaitWorkflow
         if main_workflow_returns_before_signal_completions
         else FirstCompletionCommandIsHonoredWorkflow
     )