PR Feedback 1

Author: andystaples

durabletask/entities/entity_instance_id.py

@@ -1,9 +1,7 @@
 class EntityInstanceId:
     def __init__(self, entity: str, key: str):
-        if not entity or not key:
-            raise ValueError("Entity name and key cannot be empty.")
-        if "@" in key:
-            raise ValueError("Entity key cannot contain '@' symbol.")
+        EntityInstanceId.validate_entity_name(entity)
+        EntityInstanceId.validate_key(key)
         self.entity = entity.lower()
         self.key = key
 
@@ -46,3 +44,41 @@ def parse(entity_id: str) -> "EntityInstanceId":
         except ValueError as ex:
             raise ValueError(f"Invalid entity ID format: {entity_id}") from ex
         return EntityInstanceId(entity=entity, key=key)
+
+    @staticmethod
+    def validate_entity_name(name: str) -> None:
+        """Validate that the entity name does not contain invalid characters.
+
+        Parameters
+        ----------
+        name : str
+            The entity name to validate.
+
+        Raises
+        ------
+        ValueError
+            If the name is not a valid entity name.
+        """
+        if not name:
+            raise ValueError("Entity name cannot be empty.")
+        if "@" in name:
+            raise ValueError("Entity name cannot contain '@' symbol.")
+
+    @staticmethod
+    def validate_key(key: str) -> None:
+        """Validate that the entity key does not contain invalid characters.
+
+        Parameters
+        ----------
+        key : str
+            The entity key to validate.
+
+        Raises
+        ------
+        ValueError
+            If the key is not a valid entity key.
+        """
+        if not key:
+            raise ValueError("Entity key cannot be empty.")
+        if "@" in key:
+            raise ValueError("Entity key cannot contain '@' symbol.")

durabletask/worker.py

@@ -199,9 +199,8 @@ def add_entity(self, fn: task.Entity, name: Optional[str] = None) -> str:
         return name
 
     def add_named_entity(self, name: str, fn: task.Entity) -> None:
-        if not name:
-            raise ValueError("A non-empty entity name is required.")
         name = name.lower()
+        EntityInstanceId.validate_entity_name(name)
         if name in self.entities:
             raise ValueError(f"A '{name}' entity already exists.")
 
@@ -906,7 +905,10 @@ def set_complete(
             try:
                 result_json = result if is_result_encoded else shared.to_json(result)
             except (ValueError, TypeError):
-                result_json = shared.to_json(str(JsonEncodeOutputException(result)))
+                self._is_complete = False
+                self._result = None
+                self.set_failed(JsonEncodeOutputException(result))
+                return
         action = ph.new_complete_orchestration_action(
             self.next_sequence_number(), status, result_json
         )

tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py

@@ -569,3 +569,28 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
     assert uuid.UUID(results[0]) != uuid.UUID(results[1])
     assert uuid.UUID(results[0]) != uuid.UUID(results[2])
     assert uuid.UUID(results[1]) != uuid.UUID(results[2])
+
+
+def test_orchestration_with_unparsable_output_fails():
+    def test_orchestrator(ctx: task.OrchestrationContext, _):
+        return Exception("This is not JSON serializable")
+
+    # Start a worker, which will connect to the sidecar in a background thread
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(test_orchestrator)
+        w.start()
+
+        c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                       taskhub=taskhub_name, token_credential=None)
+        id = c.schedule_new_orchestration(test_orchestrator)
+        state = c.wait_for_orchestration_completion(id, timeout=30)
+
+    assert state is not None
+    assert state.name == task.get_name(test_orchestrator)
+    assert state.instance_id == id
+    assert state.failure_details is not None
+    assert state.failure_details.error_type == "JsonEncodeOutputException"
+    assert state.failure_details.message.startswith("The orchestration result could not be encoded. Object details:")
+    assert state.failure_details.message.find("This is not JSON serializable") != -1
+    assert state.runtime_status == client.OrchestrationStatus.FAILED