Fix error in scheduler crashing everything

example/def2/obj.yaml

@@ -0,0 +1,61 @@
+---
+apiVersion: saturn.flared.io/v1alpha1
+kind: SaturnExecutor
+metadata:
+  name: default
+spec:
+  type: ProcessExecutor
+  options:
+    max_workers: 1
+    pool_type: thread
+---
+apiVersion: saturn.flared.io/v1alpha1
+kind: SaturnTopic
+metadata:
+  name: bogus-topic
+spec:
+  type: example.BogusTopic
+---
+apiVersion: saturn.flared.io/v1alpha1
+kind: SaturnTopic
+metadata:
+  name: stdout
+spec:
+  type: FileTopic
+  options:
+    path: "-"
+    mode: "w"
+---
+apiVersion: saturn.flared.io/v1alpha1
+kind: SaturnTopic
+metadata:
+  name: static-topic
+spec:
+  type: StaticTopic
+  options:
+    cycle: true
+    messages:
+    - {"id": "0", "args": {"message": "hello-0"}}
+    - {"id": "1", "args": {"message": "hello-1"}}
+    - {"id": "2", "args": {"message": "hello-2"}}
+    - {"id": "3", "args": {"message": "hello-3"}}
+---
+apiVersion: saturn.flared.io/v1alpha1
+kind: SaturnJobDefinition
+metadata:
+  name: example-job-definition
+  labels:
+    owner: team-saturn
+spec:
+  minimalInterval: "@weekly"
+  template:
+    inputs:
+      a: {topic: bogus-topic}
+      b: {topic: static-topic}
+
+    output:
+      default:
+        - topic: stdout
+
+    pipeline:
+      name: example.pipelines.fast

example/src/example/__init__.py

@@ -0,0 +1,7 @@
+from saturn_engine.worker.topics.dummy import DummyTopic
+
+
+class BogusTopic(DummyTopic):
+    async def run(self):
+        raise ValueError("Oops")
+        yield

src/saturn_engine/worker/executors/scheduler.py

@@ -111,16 +111,17 @@ async def process_task(self, task: asyncio.Task) -> AsyncIterator[T]:
             exception = task.exception()
             if exception is None:
                 yield task.result()
-            elif isinstance(exception, StopAsyncIteration):
-                self.remove(item)
             elif isinstance(exception, asyncio.CancelledError):
                 pass
+            elif isinstance(exception, Exception):
+                self.remove(item)
+                if not isinstance(exception, StopAsyncIteration):
+                    self.logger.error(
+                        "Exception raised from schedulable item",
+                        extra={"data": {"queue_name": item.name}},
+                        exc_info=exception,
+                    )
             elif exception:
-                self.logger.error(
-                    "Exception raised from schedulable item",
-                    extra={"data": {"queue_name": item.name}},
-                    exc_info=exception,
-                )
                 raise ValueError("Fatal error in schedulable") from exception
         except BaseException:
             # This is an unexpected error, likely a closed generator or

tests/worker/test_scheduler.py

@@ -73,7 +73,7 @@ async def test_scheduler(
 
 
 # TODO: Reenable once we don't cause fatal error in scheduler.
-@pytest.mark.skip
+#@pytest.mark.skip
 @pytest.mark.asyncio
 async def test_scheduler_iter_errors(scheduler: Scheduler) -> None:
     schedulable1 = make_schedulable(