fix: better reporting of errors in the head (#4640)

jina/serve/runtimes/head/__init__.py

@@ -209,7 +209,9 @@ async def process_data(self, requests: List[DataRequest], context) -> DataReques
                 else '',
                 exc_info=not self.args.quiet_error,
             )
-            raise
+            requests[0].add_exception(ex, executor=None)
+            context.set_trailing_metadata((('is-error', 'true'),))
+            return requests[0]
 
     async def process_control(self, request: ControlRequest, *args) -> ControlRequest:
         """

tests/unit/orchestrate/flow/flow-construct/test_flow_except.py

@@ -2,11 +2,11 @@
 
 import numpy as np
 import pytest
+from docarray.document.generators import from_ndarray
 
-from jina import Flow, Executor, requests, Document
+from jina import Document, Executor, Flow, requests
 from jina.excepts import RuntimeFailToStart
 from jina.proto import jina_pb2
-from docarray.document.generators import from_ndarray
 from tests import validate_callback
 
 
@@ -263,3 +263,23 @@ def test_flow_does_not_import_exec_depencies():
     with pytest.raises(RuntimeFailToStart):
         with f:
             pass
+
+
+def test_flow_head_runtime_failure(monkeypatch, capfd):
+    from jina.serve.runtimes.request_handlers.data_request_handler import (
+        DataRequestHandler,
+    )
+
+    def fail(*args, **kwargs):
+        raise NotImplementedError('Intentional error')
+
+    monkeypatch.setattr(DataRequestHandler, 'merge_routes', fail)
+
+    with Flow().add(shards=2) as f:
+        f.index(
+            [Document(text='abbcs')],
+        )
+
+    out, err = capfd.readouterr()
+    assert 'NotImplementedError' in out
+    assert 'Intentional error' in out

tests/unit/serve/runtimes/head/test_head_runtime.py

@@ -49,10 +49,10 @@ def test_control_message_processing():
     cancel_event, handle_queue, runtime_thread = _create_runtime(args)
 
     # no connection registered yet
-    with pytest.raises(RpcError):
-        GrpcConnectionPool.send_request_sync(
-            _create_test_data_message(), f'{args.host}:{args.port}'
-        )
+    resp = GrpcConnectionPool.send_request_sync(
+        _create_test_data_message(), f'{args.host}:{args.port}'
+    )
+    assert resp.status.code == resp.status.ERROR
 
     _add_worker(args, 'ip1')
     # after adding a connection, sending should work
@@ -63,10 +63,10 @@ def test_control_message_processing():
 
     _remove_worker(args, 'ip1')
     # after removing the connection again, sending does not work anymore
-    with pytest.raises(RpcError):
-        GrpcConnectionPool.send_request_sync(
-            _create_test_data_message(), f'{args.host}:{args.port}'
-        )
+    resp = GrpcConnectionPool.send_request_sync(
+        _create_test_data_message(), f'{args.host}:{args.port}'
+    )
+    assert resp.status.code == resp.status.ERROR
 
     _destroy_runtime(args, cancel_event, runtime_thread)