Handle assignment signal failures

Author: nagl-temporal

resource_pool/resource_pool_workflow.py

@@ -2,6 +2,7 @@
 from typing import Optional
 
 from temporalio import workflow
+from temporalio.exceptions import ApplicationError
 
 from resource_pool.shared import AcquireRequest, AcquireResponse
 
@@ -79,20 +80,28 @@ def get_current_holders(self) -> dict[str, Optional[InternalAcquireRequest]]:
     async def assign_resource(
         self, resource: str, internal_request: InternalAcquireRequest
     ) -> None:
-        self.resources[resource] = internal_request
         workflow.logger.info(
             f"workflow_id={internal_request.workflow_id} acquired resource {resource}"
         )
-        internal_request.release_signal = str(workflow.uuid4())
-        self.release_key_to_resource[internal_request.release_signal] = resource
 
         requester = workflow.get_external_workflow_handle(internal_request.workflow_id)
-        await requester.signal(
-            f"assign_resource_{workflow.info().workflow_id}",
-            AcquireResponse(
-                release_key=internal_request.release_signal, resource=resource
-            ),
-        )
+        try:
+            release_signal = str(workflow.uuid4())
+            await requester.signal(
+                f"assign_resource_{workflow.info().workflow_id}",
+                AcquireResponse(release_key=release_signal, resource=resource),
+            )
+
+            internal_request.release_signal = release_signal
+            self.resources[resource] = internal_request
+            self.release_key_to_resource[release_signal] = resource
+        except ApplicationError as e:
+            if e.type == "ExternalWorkflowExecutionNotFound":
+                workflow.logger.info(
+                    f"Could not assign resource {resource} to {internal_request.workflow_id}: {e.message}"
+                )
+            else:
+                raise e
 
     async def assign_next_resource(self) -> bool:
         if len(self.waiters) == 0:

tests/resource_pool/workflow_test.py

@@ -69,6 +69,13 @@ async def use_resource_mock(input: UseResourceActivityInput) -> None:
                     ), f"{workflow_id} ended on {resource} held by {holder}"
                     holder = None
 
+        # Are all the resources free, per the query?
+        pool_handle = client.get_workflow_handle_for(
+            ResourcePoolWorkflow.run, RESOURCE_POOL_WORKFLOW_ID
+        )
+        query_result = await pool_handle.query(ResourcePoolWorkflow.get_current_holders)
+        assert query_result == {"r_a": None, "r_b": None, "r_c": None}
+
 
 async def run_all_workflows(client: Client):
     resource_pool_handle = await client.start_workflow(