refactor: remove dead branches in async _pump_iterator and drop put_terminal

pyproject.toml

@@ -50,6 +50,8 @@ testpaths = ["tests"]
 asyncio_mode = "auto"
 addopts = "--import-mode=importlib"
 pythonpath = ["."]
+timeout = 10
+timeout_method = "thread"
 
 [tool.hatch.build.targets.wheel]
 packages = ["synaflow"]
@@ -83,5 +85,6 @@ dev = [
     "mkdocs-material>=9.5",
     "pytest-asyncio>=1.4.0",
     "pytest-cov>=4.0",
+    "pytest-timeout>=2.4.0",
     "ruff>=0.15.17",
 ]

synaflow/execution/async_engine/executor.py

@@ -148,43 +148,20 @@ async def _pump_iterator(
     queues: dict[str, Any],
     on_error: Any,
     dag: Dag | None = None,
-    materialize_before_enqueue: bool = False,
-    consumer_type: Any = None,
 ) -> None:
     try:
-        safe = _safe_iterate(name, iterator)
-        if materialize_before_enqueue:
-            items, _, _ = await _apply_materializer(
-                dag, name, safe, consumer_type=consumer_type
-            )
-            async for item in _safe_iterate(name, items):
-                for q in queues.values():
-                    if isinstance(q, AsyncQueueBranch):
-                        await q.put(item)
-                    else:
-                        await q.put(item)
-        else:
-            async for item in safe:
-                for q in queues.values():
-                    if isinstance(q, AsyncQueueBranch):
-                        await q.put(item)
-                    else:
-                        await q.put(item)
+        async for item in _safe_iterate(name, iterator):
+            for q in queues.values():
+                await q.put(item)
     except StepExecutionError as e:
         await _handle_error(dag, name, e.__cause__ or e)
         if on_error == OnError.STOP:
             for q in queues.values():
-                if isinstance(q, AsyncQueueBranch):
-                    await q.put_terminal(PipelineStopException(step_name=name))
-                else:
-                    await q.put(PipelineStopException(step_name=name))
+                await q.put(PipelineStopException(step_name=name))
             raise PipelineStopException(step_name=name) from e
     finally:
         for q in queues.values():
-            if isinstance(q, AsyncQueueBranch):
-                await q.put_terminal(EOF_MARKER)
-            else:
-                await q.put(EOF_MARKER)
+            await q.put(EOF_MARKER)
 
 
 async def _pump_observer(name: str, queue: asyncio.Queue, observer: Any) -> None:
@@ -737,7 +714,6 @@ async def _publish_stream_to_queues(
                 queues,
                 node.on_error,
                 dag=self.dag,
-                materialize_before_enqueue=False,
             )
         )
         self._pump_tasks.append(task)

synaflow/execution/async_engine/iterator_utils.py

@@ -31,16 +31,6 @@ async def put(self, item) -> None:
             except asyncio.QueueFull:
                 await asyncio.sleep(0.001)
 
-    async def put_terminal(self, item) -> None:
-        if not self.active:
-            return
-        while self.active:
-            try:
-                self.queue.put_nowait(item)
-                return
-            except asyncio.QueueFull:
-                await asyncio.sleep(0.001)
-
     async def get(self):
         return await self.queue.get()