Skip conditions during patch and query jobs (#137)

Fixes #129

Author: cretz

temporalio/worker/workflow_instance.py

@@ -256,15 +256,17 @@ def activate(
                     job_sets[3].append(job)
 
             # Apply every job set, running after each set
-            for job_set in job_sets:
+            for index, job_set in enumerate(job_sets):
                 if not job_set:
                     continue
                 for job in job_set:
                     # Let errors bubble out of these to the caller to fail the task
                     self._apply(job)
 
-                # Run one iteration of the loop
-                self._run_once()
+                # Run one iteration of the loop. We do not allow conditions to
+                # be checked in patch jobs (first index) or query jobs (last
+                # index).
+                self._run_once(check_conditions=index == 1 or index == 2)
         except Exception as err:
             logger.warning(
                 f"Failed activation on workflow {self._info.workflow_type} with ID {self._info.workflow_id} and run ID {self._info.run_id}",
@@ -1153,7 +1155,7 @@ def _register_task(self, task: asyncio.Task, *, name: Optional[str]) -> None:
         if hasattr(task, "_log_destroy_pending"):
             setattr(task, "_log_destroy_pending", False)
 
-    def _run_once(self) -> None:
+    def _run_once(self, *, check_conditions: bool) -> None:
         try:
             asyncio._set_running_loop(self)
 
@@ -1178,11 +1180,12 @@ def _run_once(self) -> None:
                 # Check conditions which may add to the ready list. Also remove
                 # conditions whose futures have already cancelled (e.g. when
                 # timed out).
-                self._conditions[:] = [
-                    t
-                    for t in self._conditions
-                    if not t[1].done() and not self._check_condition(*t)
-                ]
+                if check_conditions:
+                    self._conditions[:] = [
+                        t
+                        for t in self._conditions
+                        if not t[1].done() and not self._check_condition(*t)
+                    ]
         finally:
             asyncio._set_running_loop(None)
 

tests/worker/test_workflow.py

@@ -2349,6 +2349,38 @@ async def test_workflow_memo(client: Client):
             pass
 
 
+@workflow.defn
+class QueryAffectConditionWorkflow:
+    def __init__(self) -> None:
+        self.seen_query = False
+
+    @workflow.run
+    async def run(self) -> None:
+        def condition_never_after_query():
+            assert not self.seen_query
+            return False
+
+        while True:
+            await workflow.wait_condition(condition_never_after_query)
+
+    @workflow.query
+    def check_condition(self) -> bool:
+        # This is a bad thing, to mutate a workflow during a query, this is just
+        # for this test
+        self.seen_query = True
+        return True
+
+
+async def test_workflow_query_does_not_run_condition(client: Client):
+    async with new_worker(client, QueryAffectConditionWorkflow) as worker:
+        handle = await client.start_workflow(
+            QueryAffectConditionWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+        assert await handle.query(QueryAffectConditionWorkflow.check_condition)
+
+
 def new_worker(
     client: Client,
     *workflows: Type,