Handle ClientApp exception

src/py/flwr/client/app.py

@@ -18,7 +18,7 @@
 import argparse
 import sys
 import time
-from logging import DEBUG, INFO, WARN
+from logging import DEBUG, ERROR, INFO, WARN
 from pathlib import Path
 from typing import Callable, ContextManager, Optional, Tuple, Union
 
@@ -36,6 +36,8 @@
 )
 from flwr.common.logger import log, warn_deprecated_feature, warn_experimental_feature
 from flwr.common.message import Message
+from flwr.common.recordset import RecordSet
+from flwr.common.serde import message_to_taskres
 
 from .flower import load_flower_callable
 from .grpc_client.connection import grpc_connection
@@ -379,7 +381,23 @@ def _load_app() -> Flower:
                 app: Flower = load_flower_callable_fn()
 
                 # Handle task message
-                out_message = app(message=message, context=context)
+                try:
+                    out_message = app(message=message, context=context)
+                except Exception as ex:  # pylint: disable=broad-exception-caught
+                    log(ERROR, "FlowerCallable raised an exception", exc_info=ex)
+
+                    # Don't update/change RunState
+                    # Return empty Message
+
+                    error_out_message = Message(
+                        metadata=message.metadata,
+                        message=RecordSet(),
+                    )
+
+                    # Construct TaskRes from out_message
+                    error_task_res = message_to_taskres(error_out_message)
+                    send(error_task_res)
+                    continue
 
                 # Update node state
                 node_state.update_context(

src/py/flwr/driver/driver_client_proxy.py

@@ -156,8 +156,27 @@ def _send_receive_recordset(
             )
             if len(task_res_list) == 1:
                 task_res = task_res_list[0]
+                validate_task_res(task_res=task_res)
                 return serde.recordset_from_proto(task_res.task.recordset)
 
             if timeout is not None and time.time() > start_time + timeout:
                 raise RuntimeError("Timeout reached")
             time.sleep(SLEEP_TIME)
+
+
+def validate_task_res(
+    task_res: task_pb2.TaskRes,  # pylint: disable=E1101
+) -> None:
+    """Validate if a TaskRes is empty or not."""
+    if not task_res.HasField("task"):
+        raise ValueError("Invalid TaskRes, field `task` missing")
+    if not task_res.task.HasField("recordset"):
+        raise ValueError("Invalid Task, field `recordset` missing")
+
+    rs = task_res.task.recordset
+    if (
+        (not rs.parameters.keys())
+        and (not rs.metrics.keys())
+        and (not rs.configs.keys())
+    ):
+        raise ValueError("Exception during client-side task execution")

src/py/flwr/driver/driver_client_proxy_test.py

@@ -43,8 +43,13 @@
     Properties,
     Status,
 )
-from flwr.driver.driver_client_proxy import DriverClientProxy
-from flwr.proto import driver_pb2, node_pb2, task_pb2  # pylint: disable=E0611
+from flwr.driver.driver_client_proxy import DriverClientProxy, validate_task_res
+from flwr.proto import (  # pylint: disable=E0611
+    driver_pb2,
+    node_pb2,
+    recordset_pb2,
+    task_pb2,
+)
 
 MESSAGE_PARAMETERS = Parameters(tensors=[b"abc"], tensor_type="np")
 
@@ -245,3 +250,78 @@ def test_evaluate(self) -> None:
         # Assert
         assert 0.0 == evaluate_res.loss
         assert 0 == evaluate_res.num_examples
+
+    def test_validate_task_res_valid(self) -> None:
+        """Test valid TaskRes."""
+        metrics_record = recordset_pb2.MetricsRecord(  # pylint: disable=E1101
+            data={
+                "loss": recordset_pb2.MetricsRecordValue(  # pylint: disable=E1101
+                    double=1.0
+                )
+            }
+        )
+        task_res = task_pb2.TaskRes(  # pylint: disable=E1101
+            task_id="554bd3c8-8474-4b93-a7db-c7bec1bf0012",
+            group_id="",
+            run_id=0,
+            task=task_pb2.Task(  # pylint: disable=E1101
+                recordset=recordset_pb2.RecordSet(  # pylint: disable=E1101
+                    parameters={},
+                    metrics={"loss": metrics_record},
+                    configs={},
+                )
+            ),
+        )
+
+        # Execute & assert
+        try:
+            validate_task_res(task_res=task_res)
+        except ValueError:
+            self.fail()
+
+    def test_validate_task_res_missing_task(self) -> None:
+        """Test invalid TaskRes (missing task)."""
+        # Prepare
+        task_res = task_pb2.TaskRes(  # pylint: disable=E1101
+            task_id="554bd3c8-8474-4b93-a7db-c7bec1bf0012",
+            group_id="",
+            run_id=0,
+        )
+
+        # Execute & assert
+        with self.assertRaises(ValueError):
+            validate_task_res(task_res=task_res)
+
+    def test_validate_task_res_missing_recordset(self) -> None:
+        """Test invalid TaskRes (missing recordset)."""
+        # Prepare
+        task_res = task_pb2.TaskRes(  # pylint: disable=E1101
+            task_id="554bd3c8-8474-4b93-a7db-c7bec1bf0012",
+            group_id="",
+            run_id=0,
+            task=task_pb2.Task(),  # pylint: disable=E1101
+        )
+
+        # Execute & assert
+        with self.assertRaises(ValueError):
+            validate_task_res(task_res=task_res)
+
+    def test_validate_task_res_missing_content(self) -> None:
+        """Test invalid TaskRes (missing content)."""
+        # Prepare
+        task_res = task_pb2.TaskRes(  # pylint: disable=E1101
+            task_id="554bd3c8-8474-4b93-a7db-c7bec1bf0012",
+            group_id="",
+            run_id=0,
+            task=task_pb2.Task(  # pylint: disable=E1101
+                recordset=recordset_pb2.RecordSet(  # pylint: disable=E1101
+                    parameters={},
+                    metrics={},
+                    configs={},
+                )
+            ),
+        )
+
+        # Execute & assert
+        with self.assertRaises(ValueError):
+            validate_task_res(task_res=task_res)