Handle ClientApp exception

src/py/flwr/client/app.py

@@ -14,11 +14,10 @@
 # ==============================================================================
 """Flower client app."""
 
-
 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 +35,7 @@
 )
 from flwr.common.exit_handlers import register_exit_handlers
 from flwr.common.logger import log, warn_deprecated_feature, warn_experimental_feature
+from flwr.common.message import Error
 
 from .client_app import load_client_app
 from .grpc_client.connection import grpc_connection
@@ -379,7 +379,23 @@ def _load_client_app() -> ClientApp:
                 client_app: ClientApp = load_client_app_fn()
 
                 # Handle task message
-                out_message = client_app(message=message, context=context)
+                try:
+                    out_message = client_app(message=message, context=context)
+                except Exception as ex:  # pylint: disable=broad-exception-caught
+                    log(ERROR, "ClientApp raised an exception", exc_info=ex)
+
+                    # Don't update/change NodeState
+
+                    # Create error message
+                    # Reason example: "<class 'ZeroDivisionError'>:<'division by zero'>"
+                    reason = str(type(ex)) + ":<'" + str(ex) + "'>"
+                    error = Error(code=0, reason=reason)
+                    error_out_message = message.create_error_reply(error=error, ttl="")
+
+                    # Return error message
+                    send(error_out_message)
+
+                    continue
 
                 # Update node state
                 node_state.update_context(

src/py/flwr/server/compat/driver_client_proxy.py

@@ -168,8 +168,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/server/compat/driver_client_proxy_test.py

@@ -43,9 +43,13 @@
     Properties,
     Status,
 )
-from flwr.proto import driver_pb2, node_pb2, task_pb2  # pylint: disable=E0611
-
-from .driver_client_proxy import DriverClientProxy
+from flwr.proto import (  # pylint: disable=E0611
+    driver_pb2,
+    node_pb2,
+    recordset_pb2,
+    task_pb2,
+)
+from flwr.server.compat.driver_client_proxy import DriverClientProxy, validate_task_res
 
 MESSAGE_PARAMETERS = Parameters(tensors=[b"abc"], tensor_type="np")
 
@@ -248,3 +252,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)