fix race condition where pipeline exhausts with unraised errors

Author: RichardZhu2

src/pyper/_core/sync_helper/output.py

@@ -35,11 +35,11 @@ def __call__(self, *args, **kwargs):
         with ThreadPool() as tp:
             q_out = self._get_q_out(tp, *args, **kwargs)
             while True:
-                if tp.has_error:
-                    raise tp.get_error()
+                tp.raise_error_if_exists()
                 try:
                     # Use the timeout strategy for unblocking main thread without busy waiting
                     if (data := q_out.get(timeout=1)) is StopSentinel:
+                        tp.raise_error_if_exists()
                         break
                     yield data
                 except queue.Empty:

src/pyper/_core/util/thread_pool.py

@@ -35,13 +35,19 @@ def get_error(self) -> Exception:
 
     def put_error(self, e: Exception):
         self._error_queue.put(e)
+
+    def raise_error_if_exists(self):
+        if self.has_error:
+            raise self.get_error()
 
     def submit(self, func, /, args=(), kwargs=None, daemon=False):
         @functools.wraps(func)
         def wrapper(*args, **kwargs):
             try:
                 return func(*args, **kwargs)
-            except Exception as e:
+            # In the pipeline, each worker will handle errors internally, so this will never be reached in tests.
+            # This wrapper is still implemented for safety; it never makes sense not to capture errors here
+            except Exception as e:  # pragma: no cover
                 self._error_queue.put(e)
         t = threading.Thread(target=wrapper, args=args, kwargs=kwargs, daemon=daemon)
         t.start()