fix(exception): catch exception for users' runners code (#4150)

* fix(exception): raise exception if async_stream runs into exception

Signed-off-by: aarnphm-ec2-dev <29749331+aarnphm@users.noreply.github.com>

* chore: catch exceptions on all async_run

Signed-off-by: Aaron <29749331+aarnphm@users.noreply.github.com>

---------

Signed-off-by: aarnphm-ec2-dev <29749331+aarnphm@users.noreply.github.com>
Signed-off-by: Aaron <29749331+aarnphm@users.noreply.github.com>

src/bentoml/_internal/server/runner_app.py

@@ -4,6 +4,7 @@
 import json
 import logging
 import pickle
+import traceback
 import typing as t
 from typing import TYPE_CHECKING
 
@@ -205,7 +206,11 @@ async def inner():
                     # This is a workaround to allow infer stream to return a iterable of
                     # async generator, to align with how non stream inference works
                     params = paramss[0].map(AutoContainer.from_payload)
-                    ret = runner_method.async_stream(*params.args, **params.kwargs)
+                    try:
+                        ret = runner_method.async_stream(*params.args, **params.kwargs)
+                    except Exception:
+                        traceback.print_exc()
+                        raise
                     async for data in ret:
                         payload = AutoContainer.to_payload(data, 0)
                         yield payload
@@ -236,9 +241,13 @@ async def infer_batch(
                         params_list, input_batch_dim
                     )
 
-                    batch_ret = await runner_method.async_run(
-                        *batched_params.args, **batched_params.kwargs
-                    )
+                    try:
+                        batch_ret = await runner_method.async_run(
+                            *batched_params.args, **batched_params.kwargs
+                        )
+                    except Exception:
+                        traceback.print_exc()
+                        raise
 
                     # multiple output branch
                     if LazyType["tuple[t.Any, ...]"](tuple).isinstance(batch_ret):
@@ -268,7 +277,13 @@ async def infer_single(paramss: t.Sequence[Params[t.Any]]):
 
                     params = paramss[0].map(AutoContainer.from_payload)
 
-                    ret = await runner_method.async_run(*params.args, **params.kwargs)
+                    try:
+                        ret = await runner_method.async_run(
+                            *params.args, **params.kwargs
+                        )
+                    except Exception:
+                        traceback.print_exc()
+                        raise
 
                     payload = AutoContainer.to_payload(ret, 0)
                     return (payload,)