Repository: jina-ai/jina
File: jina/serve/runtimes/head/__init__.py (Python)

Code before:

        if len(worker_results) == 0:
            if exceptions:
                # raise the underlying error first
                raise exceptions[0]
            raise RuntimeError(
                f'Head {self.name} did not receive a response when sending message to worker pods'
            )

        worker_results, metadata = zip(*worker_results)

        response_request = worker_results[0]
        uses_after_metadata = None
        if self.uses_after_address:
            uses_after_result = await self.connection_pool.send_requests_once(
                worker_results,
                deployment='uses_after',
                timeout=self.timeout_send,
                retries=self._retries,
            )
            if isinstance(uses_after_result, (AioRpcError, InternalNetworkError)):
                raise uses_after_result
            else:
                (response_request, uses_after_metadata) = uses_after_result
        elif len(worker_results) > 1 and self._reduce:
            response_request = WorkerRequestHandler.reduce_requests(worker_results)
        elif len(worker_results) > 1 and not self._reduce:
            # worker returned multiple responses, but the head is configured to skip reduction
            # just concatenate the docs in this case
            response_request.data.docs = WorkerRequestHandler.get_docs_from_request(
                requests, field='docs'
            )

        merged_metadata = self._merge_metadata(
            metadata,
            uses_after_metadata,
            uses_before_metadata,
            total_shards,
            failed_shards,
        )

        return response_request, merged_metadata

    def _merge_metadata(
        self,
        metadata,
        uses_after_metadata,
        uses_before_metadata,
        total_shards,
        failed_shards,
    ):

Review comment (suggestion):
tbh since the operation modifies in place, I think it's more explicit if we don't rely on the return type and keep
WorkerRequestHandler.reduce_requests(worker_results)
I'm also fine with removing the return for the function

Code after:

        if len(worker_results) == 0:
            if exceptions:
                # raise the underlying error first
                raise exceptions[0]
            raise RuntimeError(
                f'Head {self.name} did not receive a response when sending message to worker pods'
            )

        worker_results, metadata = zip(*worker_results)

        response_request = worker_results[0]
        uses_after_metadata = None
        if self.uses_after_address:
            result = await self.connection_pool.send_requests_once(
                worker_results,
                deployment='uses_after',
                timeout=self.timeout_send,
                retries=self._retries,
            )
            if issubclass(type(result), BaseException):
                raise result
            else:
                response_request, uses_after_metadata = result
        elif len(worker_results) > 1 and self._reduce:
            response_request = WorkerRequestHandler.reduce_requests(worker_results)
        elif len(worker_results) > 1 and not self._reduce:
            # worker returned multiple responses, but the head is configured to skip reduction
            # just concatenate the docs in this case
            response_request.data.docs = WorkerRequestHandler.get_docs_from_request(
                requests, field='docs'
            )

        merged_metadata = self._merge_metadata(
            metadata,
            uses_after_metadata,
            uses_before_metadata,
            total_shards,
            failed_shards,
        )

        return response_request, merged_metadata

    def _merge_metadata(
        self,
        metadata,
        uses_after_metadata,
        uses_before_metadata,
        total_shards,
        failed_shards,
    ):