fix: expose errors occurring before the first yield of stream response

src/_bentoml_sdk/io_models.py

@@ -173,6 +173,7 @@ async def from_http_request(cls, request: Request, serde: Serde) -> IODescriptor
     @classmethod
     async def to_http_response(cls, obj: t.Any, serde: Serde) -> Response:
         """Convert a output value to HTTP response"""
+        import itertools
         import mimetypes
 
         from pydantic import RootModel
@@ -183,10 +184,24 @@ async def to_http_response(cls, obj: t.Any, serde: Serde) -> Response:
         from _bentoml_impl.serde import JSONSerde
 
         if inspect.isasyncgen(obj):
+            try:
+                # try if there is any error before the first yield
+                # if so, the error can be surfaced in the response
+                first_chunk = await obj.__anext__()
+            except StopAsyncIteration:
+                pre_chunks = []
+            else:
+                pre_chunks = [first_chunk]
+
+            async def gen() -> t.AsyncGenerator[t.Any, None]:
+                for chunk in pre_chunks:
+                    yield chunk
+                async for chunk in obj:
+                    yield chunk
 
             async def async_stream() -> t.AsyncGenerator[str | bytes, None]:
                 try:
-                    async for item in obj:
+                    async for item in gen():
                         if isinstance(item, (str, bytes)):
                             yield item
                         else:
@@ -201,6 +216,11 @@ async def async_stream() -> t.AsyncGenerator[str | bytes, None]:
             return StreamingResponse(async_stream(), media_type=cls.mime_type())
 
         elif inspect.isgenerator(obj):
+            trial, obj = itertools.tee(obj)
+            try:
+                next(trial)  # try if there is any error before the first yield
+            except StopIteration:
+                pass
 
             def content_stream() -> t.Generator[str | bytes, None, None]:
                 try:

src/bentoml/_internal/cloud/deployment.py

@@ -832,6 +832,8 @@ def is_bento_changed(bento_info: Bento) -> bool:
 
         spinner = Spinner(console=console)
         needs_update = is_bento_changed(bento_info)
+        spinner.log(f"💻 View Dashboard: {self.admin_console}")
+        endpoint_url: str | None = None
         try:
             spinner.start()
             upload_id = spinner.transmission_progress.add_task(
@@ -861,6 +863,9 @@ def is_bento_changed(bento_info: Bento) -> bool:
                     requirements_hash = self._init_deployment_files(
                         bento_dir, spinner=spinner
                     )
+                if endpoint_url is None:
+                    endpoint_url = self.get_endpoint_urls(False)[0]
+                    spinner.log(f"🌐 Endpoint: {endpoint_url}")
                 with self._tail_logs(spinner.console):
                     spinner.update("👀 Watching for changes")
                     for changes in watchfiles.watch(
@@ -937,13 +942,11 @@ def is_bento_changed(bento_info: Bento) -> bool:
                             return
         except KeyboardInterrupt:
             spinner.log(
-                "The deployment is still running, view the dashboard:\n"
-                f"    [blue]{self.admin_console}[/]\n\n"
-                "Next steps:\n"
-                "* Push the bento to BentoCloud:\n"
-                "    [blue]$ bentoml build --push[/]\n\n"
+                "\nWatcher stopped. Next steps:\n"
                 "* Attach to this deployment again:\n"
                 f"    [blue]$ bentoml develop --attach {self.name} --cluster {self.cluster}[/]\n\n"
+                "* Push the bento to BentoCloud:\n"
+                "    [blue]$ bentoml build --push[/]\n\n"
                 "* Terminate the deployment:\n"
                 f"    [blue]$ bentoml deployment terminate {self.name} --cluster {self.cluster}[/]"
             )