streamdiffusion: Report internal errors updating parameters (#796)

* live/pipelines: Reorder function definitions in iface

Initialize first small change keep going

* [POC] streamdiffusion: Report errors through logs

* Simplify error log schema

* Simplify logging schema even further

* Clean-up some error messages

* Improve _report_error output

Author: victorges

runner/app/live/pipelines/interface.py

@@ -18,6 +18,19 @@ def __init__(self):
         """Initialize pipeline with optional parameters."""
         pass
 
+    @abstractmethod
+    async def initialize(self, **params):
+        """Initialize the pipeline with parameters and warm up the processing.
+
+        This method sets up the initial pipeline state and performs warmup operations.
+        Must maintain valid state on success or restore previous state on failure.
+        Starts the pipeline loops in comfystream.
+
+        Args:
+            **params: Implementation-specific parameters
+        """
+        pass
+
     @abstractmethod
     async def put_video_frame(self, frame: VideoFrame, request_id: str):
         """Put a frame into the pipeline.
@@ -36,19 +49,6 @@ async def get_processed_video_frame(self) -> VideoOutput:
         """
         pass
 
-    @abstractmethod
-    async def initialize(self, **params):
-        """Initialize the pipeline with parameters and warm up the processing.
-
-        This method sets up the initial pipeline state and performs warmup operations.
-        Must maintain valid state on success or restore previous state on failure.
-        Starts the pipeline loops in comfystream.
-
-        Args:
-            **params: Implementation-specific parameters
-        """
-        pass
-
     @abstractmethod
     async def update_params(self, **params):
         """Update pipeline parameters.

runner/app/live/pipelines/streamdiffusion.py

@@ -106,14 +106,18 @@ async def update_params(self, **params):
                 if await self._update_params_dynamic(new_params):
                     return
             except Exception as e:
-                logging.error(f"Error updating parameters dynamically: {e}")
+                logging.error(
+                    f"[update_params] Error updating params dynamically, reloading pipeline: {e}",
+                    extra={"report_error": True},
+                    exc_info=True,
+                )
 
         logging.info(f"Resetting pipeline for params change")
 
         try:
             await self._overlay_renderer.prewarm(new_params.width, new_params.height)
         except Exception:
-            logging.debug("Failed to prewarm loading overlay caches", exc_info=True)
+            logging.warning("Failed to prewarm loading overlay caches", exc_info=True)
 
         async with self._pipeline_lock:
             # Clear the pipeline while loading the new one. The loading overlay will be shown while this is happening.
@@ -124,14 +128,18 @@ async def update_params(self, **params):
         new_pipe: Optional[StreamDiffusionWrapper] = None
         try:
             new_pipe = await asyncio.to_thread(load_streamdiffusion_sync, new_params)
-        except Exception:
-            logging.error(f"Error resetting pipeline, reloading with previous params", exc_info=True)
+        except Exception as e:
+            logging.error(
+                f"[update_params] Error reloading pipeline, falling back to previous params: {e}",
+                extra={"report_error": True},
+                exc_info=True,
+            )
             try:
                 new_params = prev_params or StreamDiffusionParams()
                 new_pipe = await asyncio.to_thread(load_streamdiffusion_sync, new_params)
-            except Exception:
-                logging.exception("Failed to reload pipeline with fallback params", stack_info=True)
-                raise
+            except Exception as e:
+                # No need to log here as we have to bubble up the error to the caller.
+                raise RuntimeError(f"Failed to reload pipeline with previous params: {e}") from e
 
         async with self._pipeline_lock:
             self.pipe = new_pipe

runner/app/live/streamer/process.py

@@ -142,7 +142,7 @@ def process_loop(self):
         try:
             asyncio.run(self._run_pipeline_loops())
         except Exception as e:
-            self._report_error(f"Error in process run method: {e}")
+            self._report_error("Error in process run method", e)
 
 
     def _handle_logging_params(self, params: dict) -> bool:
@@ -169,8 +169,7 @@ async def _initialize_pipeline(self):
                 await pipeline.initialize(**params)
                 return pipeline
         except Exception as e:
-            self._report_error(f"Error loading pipeline: {e}")
-            logging.exception(e)
+            self._report_error("Error loading pipeline", e)
             if not params:
                 # Already tried loading with default params
                 raise
@@ -182,7 +181,7 @@ async def _initialize_pipeline(self):
                     await pipeline.initialize()
                     return pipeline
             except Exception as e:
-                self._report_error(f"Error loading pipeline with default params: {e}")
+                self._report_error("Error loading pipeline with default params", e)
                 raise
 
     async def _run_pipeline_loops(self):
@@ -201,7 +200,7 @@ async def wait_for_stop():
         try:
             await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
         except Exception as e:
-            self._report_error(f"Error in pipeline loops: {e}")
+            self._report_error("Error in pipeline loops", e)
         finally:
             for task in tasks:
                 task.cancel()
@@ -223,8 +222,7 @@ async def _input_loop(self, pipeline: Pipeline):
                 # Timeout ensures the non-daemon threads from to_thread can exit if task is cancelled
                 continue
             except Exception as e:
-                self._report_error(f"Error processing input frame: {e}")
-                logging.exception(e)
+                self._report_error("Error processing input frame", e)
 
     async def _output_loop(self, pipeline: Pipeline):
         while not self.is_done():
@@ -235,7 +233,7 @@ async def _output_loop(self, pipeline: Pipeline):
                 output.log_timestamps["post_process_frame"] = time.time()
                 self._try_queue_put(self.output_queue, output)
             except Exception as e:
-                self._report_error(f"Error processing output frame: {e}")
+                self._report_error("Error processing output frame", e)
 
     async def _param_update_loop(self, pipeline: Pipeline):
         while not self.is_done():
@@ -250,7 +248,7 @@ async def _param_update_loop(self, pipeline: Pipeline):
                 with log_timing(f"PipelineProcess: Pipeline update parameters with params_hash={params_hash}"):
                     await pipeline.update_params(**params)
             except Exception as e:
-                self._report_error(f"Error updating params: {e}")
+                self._report_error("Error updating params", e)
 
     async def _get_latest_params(self, timeout: float) -> dict | None:
         """
@@ -278,12 +276,14 @@ async def _get_latest_params(self, timeout: float) -> dict | None:
 
         return params
 
-    def _report_error(self, error_msg: str):
+    def _report_error(self, msg: str, error: Exception | None = None, silent = False):
+        if not silent:
+            logging.error(msg, exc_info=error)
+
         error_event = {
-            "message": error_msg,
+            "message": f"{msg}: {error}" if error else msg,
             "timestamp": time.time()
         }
-        logging.error(error_msg)
         self._try_queue_put(self.error_queue, error_event)
 
     async def _cleanup_pipeline(self, pipeline):
@@ -356,6 +356,11 @@ def __init__(self, process: PipelineProcess):
     def emit(self, record):
         msg = self.format(record)
         self.process._try_queue_put(self.process.log_queue, msg)
+        try:
+            if getattr(record, "report_error", False):
+                self.process._report_error(record.getMessage(), silent=True)
+        except Exception as e:
+            logging.error(f"Error reporting error: {e}")
 
 # Function to clear the queue
 def clear_queue(queue):