Repository: jina-ai/serve
Pull request: feat: add multiple attempts options to client.post API
Goal of this PR is to enable `retry_mechanism` in the client.

- [x] Expose `max_attempts`, `initial_backoff`, `max_backoff` and `backoff_multiplier` in the `client.post` API
- [x] Clarify usage of `continue_on_error` with any kind of Errors
- [ ] Test features
- [x] Document

**Changes**
- Allow to pass `max_attempts`, `initial_backoff`, `max_backoff` and `backoff_multiplier` to client.post API
   - In gRPC, these are added as configuration to the grpcChannel
   - In HTTP and websocket, we do a manual retry following the formula proposed by grpc
```The initial retry attempt will occur at random(0, initialBackoff). In general, the n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff)).```

- Document `continue_on_error` behavior.

Problem statement:
fix: no retries in case of network problems
Hi, can we have retries for indexing documents at the flow?
If I disconnect from the internet during indexing and then connect again after 5 seconds, the flow fails without doing retries.
This issue slows us down when making deployments.

Suggested interface:

```python
f.index(inputs=docs, request_size=10, retries=3, retry_delay=10s)
```
In case a request is not successful even after the retry, the program execution should continue.
Having not complete data is much better for us than completely failing.

```
▶ indexing 67354 documents
⠴ Working... ━━━━━━━━━━━━╺━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0:37:44  30% ETA: 1:24:28  
Traceback (most recent call last):
  File "/Users/florianhonicke/jina/now/now/admin/deploy_text_to_video.py", line 39, in <module>
    result = deploy()
  File "/Users/florianhonicke/jina/now/now/admin/deploy_text_to_video.py", line 34, in deploy
    response = cli(args=kwargs)
  File "/Users/florianhonicke/jina/now/now/cli/__init__.py", line 143, in cli
    return run_k8s(os_type=os_type, arch=arch, **args)
  File "/Users/florianhonicke/jina/now/now/run_all_k8s.py", line 168, in run_k8s
    **kwargs,
  File "/Users/florianhonicke/jina/now/now/run_all_k8s.py", line 111, in start_now
    ) = run_backend.run(app_instance, user_input, kubectl_path=kwargs['kubectl_path'])
  File "/Users/florianhonicke/jina/now/now/log/log.py", line 58, in profiled_fun
    result = fun(*args, **kwargs)
  File "/Users/florianhonicke/jina/now/now/run_backend.py", line 62, in run
    return_results=False,
  File "/Users/florianhonicke/jina/now/now/log/log.py", line 58, in profiled_fun
    result = fun(*args, **kwargs)
  File "/Users/florianhonicke/jina/now/now/run_backend.py", line 101, in call_flow
    return_results=return_results,
  File "/Users/florianhonicke/jina/now/env/lib/python3.7/site-packages/jina/clients/mixin.py", line 271, in post
    **kwargs,
  File "/Users/florianhonicke/jina/now/env/lib/python3.7/site-packages/jina/helper.py", line 1329, in run_async
    return asyncio.run(func(*args, **kwargs))
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "uvloop/loop.pyx", line 1501, in uvloop.loop.Loop.run_until_complete
  File "/Users/florianhonicke/jina/now/env/lib/python3.7/site-packages/jina/clients/mixin.py", line 252, in _get_results
    async for resp in c._get_results(*args, **kwargs):
  File "/Users/florianhonicke/jina/now/env/lib/python3.7/site-packages/jina/clients/base/grpc.py", line 119, in _get_results
    raise ConnectionError(my_details) from None
ConnectionError: Operation timed out
```

Document `continue_on_error` parameter
Add Documentation about `continue_on_error` parameter

Hints from issue discussion:
Hey @florian-hoenicke ,

To make sure that the `indexing` continues, please use the `continue_on_error` parameter from the `client.post`.

Commit message: fix: apply suggestions from code review

Co-authored-by: Johannes Messner <44071807+JohannesMessner@users.noreply.github.com>

Patch to review:
diff --git a/docs/fundamentals/client/client.md b/docs/fundamentals/client/client.md
index cc4d63bfe2163..5994e9baf1649 100644
--- a/docs/fundamentals/client/client.md
+++ b/docs/fundamentals/client/client.md
@@ -479,14 +479,9 @@ For example, a `SIGKILL` from the client OS during the handling of the request,
 will not trigger the callback.
 
 
-
 Callback functions in Jina expect a `Response` of the type {class}`~jina.types.request.data.DataRequest`, which contains resulting Documents,
 parameters, and other information.
 
-
-
-
-
 ### Handle DataRequest in callbacks
 
 `DataRequest`s are objects that are sent by Jina internally. Callback functions process DataRequests, and `client.post()`
@@ -668,7 +663,26 @@ with Flow().add() as f, open('output.txt', 'w') as fp:
     )
 ```
 
+````{admonition} What errors can be handled by the callback
+:class: caution
+Callback can handle errors that are caused by Executors raising an Exception.
+It will not receive exceptions or from the Gateway having connectivity errors with the Executors or between the Client and the Gateway.
+````
 
+### Continue streaming when an error occurs
+
+`client.post()` accepts a `continue_on_error` parameter. When set to `True`, the Client will keep trying to send the remaining requests. The `continue_on_error` parameter will only apply
+to Exceptions caused by an Executor, but in case of network connectivity issues, an Exception will be raised.
+
+### Transient fault handling with retries
+
+`client.post()` accepts `max_attempts`, `initial_backoff`, `max_backoff` and `backoff_multiplier` parameters to control the capacity to retry requests, when a transient connectivity error occurs, using an exponential backoff strategy, 
+This can help to overcome transient network connectivity issues. 
+
+The `max_attempts` determines the number of sending attempts, including the original request.
+The `initial_backoff`, `max_backoff`, and `backoff_multiplier` parameters determine the randomized delay in seconds before retry attempts.
+
+The initial retry attempt will occur at random(0, initial_backoff). In general, the n-th attempt will occur at random(0, min(initial_backoff*backoff_multiplier**(n-1), max_backoff)).
 
 ## Returns
 
@@ -809,4 +823,4 @@ For details on the allowed mutation arguments and response fields, see {ref}`her
 :hidden:
 
 third-party-client
-```
\ No newline at end of file
+```
diff --git a/jina/clients/base/grpc.py b/jina/clients/base/grpc.py
index 6f614d517fe44..174c351b47788 100644
--- a/jina/clients/base/grpc.py
+++ b/jina/clients/base/grpc.py
@@ -2,6 +2,7 @@
 from typing import TYPE_CHECKING, Optional
 
 import grpc
+import json
 
 from jina.clients.base import BaseClient
 from jina.clients.helper import callback_exec
@@ -62,6 +63,10 @@ async def _get_results(
         on_error: Optional['CallbackFnType'] = None,
         on_always: Optional['CallbackFnType'] = None,
         compression: Optional[str] = None,
+        max_attempts: int = 1,
+        initial_backoff: float = 0.5,
+        max_backoff: float = 0.1,
+        backoff_multiplier: float = 1.5,
         **kwargs,
     ):
         try:
@@ -73,8 +78,30 @@ async def _get_results(
 
             self.inputs = inputs
             req_iter = self._get_requests(**kwargs)
+            continue_on_error = self.continue_on_error
+            # while loop with retries, check in which state the `iterator` remains after failure
+            options = GrpcConnectionPool.get_default_grpc_options()
+            if max_attempts > 1:
+                service_config_json = json.dumps({
+                                "methodConfig": [{
+                                    # To apply retry to all methods, put [{}] in the "name" field
+                                    "name": [{}],
+                                    "retryPolicy": {
+                                        "maxAttempts": max_attempts,
+                                        "initialBackoff": f"{initial_backoff}s",
+                                        "maxBackoff": f"{max_backoff}s",
+                                        "backoffMultiplier": {backoff_multiplier},
+                                        "retryableStatusCodes": ["UNAVAILABLE", "DEADLINE_EXCEEDED", "INTERNAL"],
+                                    },
+                                }]
+                            })
+                # NOTE: the retry feature will be enabled by default >=v1.40.0
+                options.append(("grpc.enable_retries", 1))
+                options.append(("grpc.service_config", service_config_json))
+
             async with GrpcConnectionPool.get_grpc_channel(
                 f'{self.args.host}:{self.args.port}',
+                options=options,
                 asyncio=True,
                 tls=self.args.tls,
             ) as channel:
@@ -84,58 +111,59 @@ async def _get_results(
                 with ProgressBar(
                     total_length=self._inputs_length, disable=not self.show_progress
                 ) as p_bar:
-                    async for resp in stub.Call(
-                        req_iter,
-                        compression=self.compression,
-                        metadata=kwargs.get('metadata', None),
-                        credentials=kwargs.get('credentials', None),
-                        timeout=kwargs.get('timeout', None),
-                    ):
-                        callback_exec(
-                            response=resp,
-                            on_error=on_error,
-                            on_done=on_done,
-                            on_always=on_always,
-                            continue_on_error=self.continue_on_error,
-                            logger=self.logger,
-                        )
-                        if self.show_progress:
-                            p_bar.update()
-                        yield resp
+                    try:
+                        async for resp in stub.Call(
+                            req_iter,
+                            compression=self.compression,
+                            metadata=kwargs.get('metadata', None),
+                            credentials=kwargs.get('credentials', None),
+                            timeout=kwargs.get('timeout', None),
+                        ):
+                            callback_exec(
+                                response=resp,
+                                on_error=on_error,
+                                on_done=on_done,
+                                on_always=on_always,
+                                continue_on_error=continue_on_error,
+                                logger=self.logger,
+                            )
+                            if self.show_progress:
+                                p_bar.update()
+                            yield resp
+                    except (grpc.aio._call.AioRpcError, InternalNetworkError) as err:
+                        my_code = err.code()
+                        my_details = err.details()
+                        msg = f'gRPC error: {my_code} {my_details}'
+
+                        if my_code == grpc.StatusCode.UNAVAILABLE:
+                            self.logger.error(
+                                f'{msg}\nThe ongoing request is terminated as the server is not available or closed already.'
+                            )
+                            raise ConnectionError(my_details)
+                        elif my_code == grpc.StatusCode.DEADLINE_EXCEEDED:
+                            self.logger.error(
+                                f'{msg}\nThe ongoing request is terminated due to a server-side timeout.'
+                            )
+                            raise ConnectionError(my_details)
+                        elif my_code == grpc.StatusCode.INTERNAL:
+                            self.logger.error(f'{msg}\ninternal error on the server side')
+                            raise err
+                        elif (
+                                my_code == grpc.StatusCode.UNKNOWN
+                                and 'asyncio.exceptions.TimeoutError' in my_details
+                        ):
+                            raise BadClientInput(
+                                f'{msg}\n'
+                                'often the case is that you define/send a bad input iterator to jina, '
+                                'please double check your input iterator'
+                            ) from err
+                        else:
+                            raise BadServerFlow(msg) from err
 
         except KeyboardInterrupt:
             self.logger.warning('user cancel the process')
         except asyncio.CancelledError as ex:
             self.logger.warning(f'process error: {ex!r}')
-        except (grpc.aio._call.AioRpcError, InternalNetworkError) as err:
-            my_code = err.code()
-            my_details = err.details()
-            msg = f'gRPC error: {my_code} {my_details}'
-
-            if my_code == grpc.StatusCode.UNAVAILABLE:
-                self.logger.error(
-                    f'{msg}\nThe ongoing request is terminated as the server is not available or closed already.'
-                )
-                raise ConnectionError(my_details) from None
-            elif my_code == grpc.StatusCode.DEADLINE_EXCEEDED:
-                self.logger.error(
-                    f'{msg}\nThe ongoing request is terminated due to a server-side timeout.'
-                )
-                raise ConnectionError(my_details) from None
-            elif my_code == grpc.StatusCode.INTERNAL:
-                self.logger.error(f'{msg}\ninternal error on the server side')
-                raise err
-            elif (
-                my_code == grpc.StatusCode.UNKNOWN
-                and 'asyncio.exceptions.TimeoutError' in my_details
-            ):
-                raise BadClientInput(
-                    f'{msg}\n'
-                    'often the case is that you define/send a bad input iterator to jina, '
-                    'please double check your input iterator'
-                ) from err
-            else:
-                raise BadServerFlow(msg) from err
         except:
             # Not sure why, adding this line helps in fixing a hanging test
             raise
diff --git a/jina/clients/base/helper.py b/jina/clients/base/helper.py
index 2623e2e51955c..4532aa2fc0ad7 100644
--- a/jina/clients/base/helper.py
+++ b/jina/clients/base/helper.py
@@ -1,4 +1,5 @@
 import asyncio
+import random
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
@@ -17,11 +18,21 @@
 class AioHttpClientlet(ABC):
     """aiohttp session manager"""
 
-    def __init__(self, url: str, logger: 'JinaLogger', **kwargs) -> None:
+    def __init__(self, url: str,
+                 logger: 'JinaLogger',
+                 max_attempts: int = 1,
+                 initial_backoff: float = 0.5,
+                 max_backoff: float = 0.1,
+                 backoff_multiplier: float = 1.5,
+                 **kwargs) -> None:
         """HTTP Client to be used with the streamer
 
         :param url: url to send http/websocket request to
         :param logger: jina logger
+        :param max_attempts: Number of sending attempts, including the original request.
+        :param initial_backoff: The first retry will happen with a delay of random(0, initial_backoff)
+        :param max_backoff: The maximum accepted backoff after the exponential incremental delay
+        :param backoff_multiplier: The n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff))
         :param kwargs: kwargs  which will be forwarded to the `aiohttp.Session` instance. Used to pass headers to requests
         """
         self.url = url
@@ -36,6 +47,10 @@ def __init__(self, url: str, logger: 'JinaLogger', **kwargs) -> None:
             self._session_kwargs['auth'] = kwargs.get('auth')
         if kwargs.get('cookies', None):
             self._session_kwargs['cookies'] = kwargs.get('cookies')
+        self.max_attempts = max_attempts
+        self.initial_backoff = initial_backoff
+        self.max_backoff = max_backoff
+        self.backoff_multiplier = backoff_multiplier
 
     @abstractmethod
     async def send_message(self, **kwargs):
@@ -103,7 +118,15 @@ async def send_message(self, request: 'Request'):
         req_dict['exec_endpoint'] = req_dict['header']['exec_endpoint']
         if 'target_executor' in req_dict['header']:
             req_dict['target_executor'] = req_dict['header']['target_executor']
-        return await self.session.post(url=self.url, json=req_dict).__aenter__()
+        for retry in range(1, self.max_attempts + 1):
+            try:
+                return await self.session.post(url=self.url, json=req_dict).__aenter__()
+            except:
+                if retry == self.max_attempts:
+                    raise
+                else:
+                    wait_time = random.uniform(0, min(self.initial_backoff*self.backoff_multiplier**(retry-1), self.max_backoff))
+                    await asyncio.sleep(wait_time)
 
     async def send_dry_run(self):
         """Query the dry_run endpoint from Gateway
@@ -161,10 +184,17 @@ async def send_message(self, request: 'Request'):
         :param request: request object
         :return: send request as bytes awaitable
         """
-        try:
-            return await self.websocket.send_bytes(request.SerializeToString())
-        except ConnectionResetError:
-            self.logger.critical(f'server connection closed already!')
+        # we have to send here `retries` information
+        for retry in range(1, self.max_attempts + 1):
+            try:
+                return await self.websocket.send_bytes(request.SerializeToString())
+            except ConnectionResetError:
+                if retry == self.max_attempts:
+                    self.logger.critical(f'server connection closed already!')
+                    raise
+                else:
+                    wait_time = random.uniform(0, min(self.initial_backoff*self.backoff_multiplier**(retry-1), self.max_backoff))
+                    await asyncio.sleep(wait_time)
 
     async def send_dry_run(self):
         """Query the dry_run endpoint from Gateway
diff --git a/jina/clients/base/http.py b/jina/clients/base/http.py
index ab1d67e4188b1..b78c033d0716a 100644
--- a/jina/clients/base/http.py
+++ b/jina/clients/base/http.py
@@ -25,19 +25,19 @@ def _handle_response_status(self, r_status, r_str, url):
         if r_status == status.HTTP_404_NOT_FOUND:
             raise BadClient(f'no such endpoint {url}')
         elif (
-            r_status == status.HTTP_503_SERVICE_UNAVAILABLE
-            or r_status == status.HTTP_504_GATEWAY_TIMEOUT
+                r_status == status.HTTP_503_SERVICE_UNAVAILABLE
+                or r_status == status.HTTP_504_GATEWAY_TIMEOUT
         ):
             if (
-                'header' in r_str
-                and 'status' in r_str['header']
-                and 'description' in r_str['header']['status']
+                    'header' in r_str
+                    and 'status' in r_str['header']
+                    and 'description' in r_str['header']['status']
             ):
                 raise ConnectionError(r_str['header']['status']['description'])
             else:
                 raise ValueError(r_str)
         elif (
-            r_status < status.HTTP_200_OK or r_status > status.HTTP_300_MULTIPLE_CHOICES
+                r_status < status.HTTP_200_OK or r_status > status.HTTP_300_MULTIPLE_CHOICES
         ):  # failure codes
             raise ValueError(r_str)
 
@@ -75,18 +75,26 @@ async def _is_flow_ready(self, **kwargs) -> bool:
         return False
 
     async def _get_results(
-        self,
-        inputs: 'InputType',
-        on_done: 'CallbackFnType',
-        on_error: Optional['CallbackFnType'] = None,
-        on_always: Optional['CallbackFnType'] = None,
-        **kwargs,
+            self,
+            inputs: 'InputType',
+            on_done: 'CallbackFnType',
+            on_error: Optional['CallbackFnType'] = None,
+            on_always: Optional['CallbackFnType'] = None,
+            max_attempts: int = 1,
+            initial_backoff: float = 0.5,
+            max_backoff: float = 0.1,
+            backoff_multiplier: float = 1.5,
+            **kwargs,
     ):
         """
         :param inputs: the callable
         :param on_done: the callback for on_done
         :param on_error: the callback for on_error
         :param on_always: the callback for on_always
+        :param max_attempts: Number of sending attempts, including the original request.
+        :param initial_backoff: The first retry will happen with a delay of random(0, initial_backoff)
+        :param max_backoff: The maximum accepted backoff after the exponential incremental delay
+        :param backoff_multiplier: The n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff))
         :param kwargs: kwargs coming from the public interface. Includes arguments to be passed to the `HTTPClientlet`
         :yields: generator over results
         """
@@ -105,11 +113,12 @@ async def _get_results(
             proto = 'https' if self.args.tls else 'http'
             url = f'{proto}://{self.args.host}:{self.args.port}/post'
             iolet = await stack.enter_async_context(
-                HTTPClientlet(url=url, logger=self.logger, **kwargs)
+                HTTPClientlet(url=url, logger=self.logger, max_attempts=max_attempts, initial_backoff=initial_backoff,
+                              max_backoff=max_backoff, backoff_multiplier=backoff_multiplier, **kwargs)
             )
 
             def _request_handler(
-                request: 'Request',
+                    request: 'Request',
             ) -> 'Tuple[asyncio.Future, Optional[asyncio.Future]]':
                 """
                 For HTTP Client, for each request in the iterator, we `send_message` using
diff --git a/jina/clients/base/websocket.py b/jina/clients/base/websocket.py
index 9ddf8d8285fcd..49323e2d1a351 100644
--- a/jina/clients/base/websocket.py
+++ b/jina/clients/base/websocket.py
@@ -77,6 +77,10 @@ async def _get_results(
         on_done: 'CallbackFnType',
         on_error: Optional['CallbackFnType'] = None,
         on_always: Optional['CallbackFnType'] = None,
+        max_attempts: int = 1,
+        initial_backoff: float = 0.5,
+        max_backoff: float = 0.1,
+        backoff_multiplier: float = 1.5,
         **kwargs,
     ):
         """
@@ -84,6 +88,10 @@ async def _get_results(
         :param on_done: the callback for on_done
         :param on_error: the callback for on_error
         :param on_always: the callback for on_always
+        :param max_attempts: Number of sending attempts, including the original request.
+        :param initial_backoff: The first retry will happen with a delay of random(0, initial_backoff)
+        :param max_backoff: The maximum accepted backoff after the exponential incremental delay
+        :param backoff_multiplier: The n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff))
         :param kwargs: kwargs coming from the public interface. Includes arguments to be passed to the `WebsocketClientlet`
         :yields: generator over results
         """
@@ -102,7 +110,8 @@ async def _get_results(
             proto = 'wss' if self.args.tls else 'ws'
             url = f'{proto}://{self.args.host}:{self.args.port}/'
             iolet = await stack.enter_async_context(
-                WebsocketClientlet(url=url, logger=self.logger, **kwargs)
+                WebsocketClientlet(url=url, logger=self.logger, max_attempts=max_attempts, initial_backoff=initial_backoff,
+                                   max_backoff=max_backoff, backoff_multiplier=backoff_multiplier, **kwargs)
             )
 
             request_buffer: Dict[
diff --git a/jina/clients/mixin.py b/jina/clients/mixin.py
index 719cf2ee75a8b..95d2e938a6054 100644
--- a/jina/clients/mixin.py
+++ b/jina/clients/mixin.py
@@ -35,11 +35,11 @@ class MutateMixin:
     """The GraphQL Mutation Mixin for Client and Flow"""
 
     def mutate(
-        self,
-        mutation: str,
-        variables: Optional[dict] = None,
-        timeout: Optional[float] = None,
-        headers: Optional[dict] = None,
+            self,
+            mutation: str,
+            variables: Optional[dict] = None,
+            timeout: Optional[float] = None,
+            headers: Optional[dict] = None,
     ):
         """Perform a GraphQL mutation
 
@@ -70,11 +70,11 @@ class AsyncMutateMixin(MutateMixin):
     """The async GraphQL Mutation Mixin for Client and Flow"""
 
     async def mutate(
-        self,
-        mutation: str,
-        variables: Optional[dict] = None,
-        timeout: Optional[float] = None,
-        headers: Optional[dict] = None,
+            self,
+            mutation: str,
+            variables: Optional[dict] = None,
+            timeout: Optional[float] = None,
+            headers: Optional[dict] = None,
     ):
         """Perform a GraphQL mutation, asynchronously
 
@@ -118,20 +118,19 @@ async def is_flow_ready(self, **kwargs) -> bool:
 
 
 def _render_response_table(r, st, ed, show_table: bool = True):
-
     from rich import print
 
     elapsed = (ed - st) * 1000
     route = r.routes
     gateway_time = (
-        route[0].end_time.ToMilliseconds() - route[0].start_time.ToMilliseconds()
+            route[0].end_time.ToMilliseconds() - route[0].start_time.ToMilliseconds()
     )
     exec_time = {}
 
     if len(route) > 1:
         for r in route[1:]:
             exec_time[r.executor] = (
-                r.end_time.ToMilliseconds() - r.start_time.ToMilliseconds()
+                    r.end_time.ToMilliseconds() - r.start_time.ToMilliseconds()
             )
     network_time = elapsed - gateway_time
     server_network = gateway_time - sum(exec_time.values())
@@ -206,19 +205,23 @@ class PostMixin:
     """The Post Mixin class for Client and Flow"""
 
     def post(
-        self,
-        on: str,
-        inputs: Optional['InputType'] = None,
-        on_done: Optional['CallbackFnType'] = None,
-        on_error: Optional['CallbackFnType'] = None,
-        on_always: Optional['CallbackFnType'] = None,
-        parameters: Optional[Dict] = None,
-        target_executor: Optional[str] = None,
-        request_size: int = 100,
-        show_progress: bool = False,
-        continue_on_error: bool = False,
-        return_responses: bool = False,
-        **kwargs,
+            self,
+            on: str,
+            inputs: Optional['InputType'] = None,
+            on_done: Optional['CallbackFnType'] = None,
+            on_error: Optional['CallbackFnType'] = None,
+            on_always: Optional['CallbackFnType'] = None,
+            parameters: Optional[Dict] = None,
+            target_executor: Optional[str] = None,
+            request_size: int = 100,
+            show_progress: bool = False,
+            continue_on_error: bool = False,
+            return_responses: bool = False,
+            max_attempts: int = 1,
+            initial_backoff: float = 0.5,
+            max_backoff: float = 0.1,
+            backoff_multiplier: float = 1.5,
+            **kwargs,
     ) -> Optional[Union['DocumentArray', List['Response']]]:
         """Post a general data request to the Flow.
 
@@ -231,9 +234,12 @@ def post(
         :param target_executor: a regex string. Only matching Executors will process the request.
         :param request_size: the number of Documents per request. <=0 means all inputs in one request.
         :param show_progress: if set, client will show a progress bar on receiving every request.
-        :param continue_on_error: if set, a Request that causes callback error will be logged only without blocking the further requests.7
+        :param continue_on_error: if set, a Request that causes an error will be logged only without blocking the further requests.
         :param return_responses: if set to True, the result will come as Response and not as a `DocumentArray`
-
+        :param max_attempts: Number of sending attempts, including the original request.
+        :param initial_backoff: The first retry will happen with a delay of random(0, initial_backoff)
+        :param max_backoff: The maximum accepted backoff after the exponential incremental delay
+        :param backoff_multiplier: The n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff))
         :param kwargs: additional parameters
         :return: None or DocumentArray containing all response Documents
 
@@ -272,6 +278,10 @@ async def _get_results(*args, **kwargs):
             target_executor=target_executor,
             parameters=parameters,
             request_size=request_size,
+            max_attempts=max_attempts,
+            initial_backoff=initial_backoff,
+            max_backoff=max_backoff,
+            backoff_multiplier=backoff_multiplier,
             **kwargs,
         )
 
@@ -286,19 +296,23 @@ class AsyncPostMixin:
     """The Async Post Mixin class for AsyncClient and AsyncFlow"""
 
     async def post(
-        self,
-        on: str,
-        inputs: Optional['InputType'] = None,
-        on_done: Optional['CallbackFnType'] = None,
-        on_error: Optional['CallbackFnType'] = None,
-        on_always: Optional['CallbackFnType'] = None,
-        parameters: Optional[Dict] = None,
-        target_executor: Optional[str] = None,
-        request_size: int = 100,
-        show_progress: bool = False,
-        continue_on_error: bool = False,
-        return_responses: bool = False,
-        **kwargs,
+            self,
+            on: str,
+            inputs: Optional['InputType'] = None,
+            on_done: Optional['CallbackFnType'] = None,
+            on_error: Optional['CallbackFnType'] = None,
+            on_always: Optional['CallbackFnType'] = None,
+            parameters: Optional[Dict] = None,
+            target_executor: Optional[str] = None,
+            request_size: int = 100,
+            show_progress: bool = False,
+            continue_on_error: bool = False,
+            return_responses: bool = False,
+            max_attempts: int = 1,
+            initial_backoff: float = 0.5,
+            max_backoff: float = 0.1,
+            backoff_multiplier: float = 1.5,
+            **kwargs,
     ) -> AsyncGenerator[None, Union['DocumentArray', 'Response']]:
         """Async Post a general data request to the Flow.
 
@@ -311,8 +325,12 @@ async def post(
         :param target_executor: a regex string. Only matching Executors will process the request.
         :param request_size: the number of Documents per request. <=0 means all inputs in one request.
         :param show_progress: if set, client will show a progress bar on receiving every request.
-        :param continue_on_error: if set, a Request that causes callback error will be logged only without blocking the further requests.
+        :param continue_on_error: if set, a Request that causes an error will be logged only without blocking the further requests.
         :param return_responses: if set to True, the result will come as Response and not as a `DocumentArray`
+        :param max_attempts: Number of sending attempts, including the original request.
+        :param initial_backoff: The first retry will happen with a delay of random(0, initial_backoff)
+        :param max_backoff: The maximum accepted backoff after the exponential incremental delay
+        :param backoff_multiplier: The n-th attempt will occur at random(0, min(initialBackoff*backoffMultiplier**(n-1), maxBackoff))
         :param kwargs: additional parameters, can be used to pass metadata or authentication information in the server call
         :yield: Response object
 
@@ -326,15 +344,19 @@ async def post(
         parameters = _include_results_field_in_param(parameters)
 
         async for result in c._get_results(
-            inputs=inputs,
-            on_done=on_done,
-            on_error=on_error,
-            on_always=on_always,
-            exec_endpoint=on,
-            target_executor=target_executor,
-            parameters=parameters,
-            request_size=request_size,
-            **kwargs,
+                inputs=inputs,
+                on_done=on_done,
+                on_error=on_error,
+                on_always=on_always,
+                exec_endpoint=on,
+                target_executor=target_executor,
+                parameters=parameters,
+                request_size=request_size,
+                max_attempts=max_attempts,
+                initial_backoff=initial_backoff,
+                max_backoff=max_backoff,
+                backoff_multiplier=backoff_multiplier,
+                **kwargs,
         ):
             if not return_responses:
                 yield result.data.docs


Code review comments:
diff_hunk: @@ -668,7 +663,26 @@ with Flow().add() as f, open('output.txt', 'w') as fp:
     )
 ```
 
+````{admonition} What errors can be handled by the callback
+:class: caution
+Callback can handle errors that are caused by Executors raising an Exception.
+It will not receive exceptions or from the Gateway having connectivity errors with the Executors or between the Client and the Gateway.
path: docs/fundamentals/client/client.md
text: @user1:
```suggestion
Callbacks can handle errors that are caused by Executors raising an Exception.

A callback will not receive exceptions:
- from the Gateway having connectivity errors with the Executors.
- between the Client and the Gateway.
```

I'm not clear on the "or from" part here. Is this a typo? Regardless, I've changed it as best I can. Does the admonition box support bullets?