Handle Serialization Errors in List Steps (#478)

Author: kraftp

dbos/_debouncer.py

@@ -86,6 +86,7 @@ def debouncer_workflow(
     dbos = _get_dbos_instance()
 
     workflow_inputs: WorkflowInputs = {"args": args, "kwargs": kwargs}
+
     # Every time the debounced workflow is called, a message is sent to this workflow.
     # It waits until debounce_period_sec have passed since the last message or until
     # debounce_timeout_sec has elapsed.
@@ -95,7 +96,10 @@ def get_debounce_deadline_epoch_sec() -> float:
             if options["debounce_timeout_sec"]
             else math.inf
         )
-    debounce_deadline_epoch_sec = dbos._sys_db.call_function_as_step(get_debounce_deadline_epoch_sec, "get_debounce_deadline_epoch_sec")
+
+    debounce_deadline_epoch_sec = dbos._sys_db.call_function_as_step(
+        get_debounce_deadline_epoch_sec, "get_debounce_deadline_epoch_sec"
+    )
     debounce_period_sec = initial_debounce_period_sec
     while time.time() < debounce_deadline_epoch_sec:
         time_until_deadline = max(debounce_deadline_epoch_sec - time.time(), 0)

dbos/_sys_db.py

@@ -1077,24 +1077,23 @@ def get_workflow_steps(self, workflow_id: str) -> List[StepInfo]:
                     SystemSchema.operation_outputs.c.child_workflow_id,
                 ).where(SystemSchema.operation_outputs.c.workflow_uuid == workflow_id)
             ).fetchall()
-            return [
-                StepInfo(
+            steps = []
+            for row in rows:
+                _, output, exception = _serialization.safe_deserialize(
+                    workflow_id,
+                    serialized_input=None,
+                    serialized_output=row[2],
+                    serialized_exception=row[3],
+                )
+                step = StepInfo(
                     function_id=row[0],
                     function_name=row[1],
-                    output=(
-                        _serialization.deserialize(row[2])
-                        if row[2] is not None
-                        else row[2]
-                    ),
-                    error=(
-                        _serialization.deserialize_exception(row[3])
-                        if row[3] is not None
-                        else row[3]
-                    ),
+                    output=output,
+                    error=exception,
                     child_workflow_id=row[4],
                 )
-                for row in rows
-            ]
+                steps.append(step)
+            return steps
 
     def _record_operation_result_txn(
         self, result: OperationResultInternal, conn: sa.Connection

tests/test_failures.py

@@ -9,6 +9,7 @@
 from sqlalchemy.exc import InvalidRequestError, OperationalError
 
 from dbos import DBOS, Queue, SetWorkflowID
+from dbos._client import DBOSClient
 from dbos._dbos_config import DBOSConfig
 from dbos._error import (
     DBOSAwaitedWorkflowCancelledError,
@@ -502,6 +503,36 @@ def test_error_serialization() -> None:
     assert isinstance(exception, str)
 
 
+def test_workflow_error_serialization(dbos: DBOS, client: DBOSClient) -> None:
+
+    @DBOS.step()
+    def step() -> None:
+        raise BadException(1, 2)
+
+    @DBOS.workflow()
+    def workflow() -> None:
+        step()
+
+    handle = DBOS.start_workflow(workflow)
+
+    with pytest.raises(BadException):
+        handle.get_result()
+
+    workflows = DBOS.list_workflows()
+    assert len(workflows) == 1
+    assert workflows[0].error is not None
+
+    steps = DBOS.list_workflow_steps(handle.workflow_id)
+    assert len(steps) == 1
+    assert steps[0]["error"] is not None
+
+    status = handle.get_status()
+    assert status.error is not None
+
+    status = client.retrieve_workflow(handle.workflow_id).get_status()
+    assert status.error is not None
+
+
 def test_unregistered_workflow(dbos: DBOS, config: DBOSConfig) -> None:
 
     @DBOS.workflow()