pause_intake answers False when claimed work outlasts its timeout, and reload then resumes intake and raises RebuildBusy before tearing anything down

backend/fluksio/flow/controller.py

@@ -741,8 +741,13 @@ class FlowController:
         try:
             # Work already claimed belongs to the pipeline it was claimed
             # against; let it finish there before swapping the graph out.
-            if self.execution is not None:
-                await run_in_threadpool(self.execution.pause_intake)
+            # Work that does not finish in time keeps the old graph, which
+            # nothing has touched yet.
+            if self.execution is not None and not await run_in_threadpool(
+                self.execution.pause_intake
+            ):
+                self.execution.resume_intake()
+                raise RebuildBusy("Claimed work is still running on the pipeline")
             await self._teardown()
             # A fresh supervisor per build, so a flow quarantined by the last
             # one gets another chance once its author has changed something.

backend/fluksio/flow/executor.py

@@ -255,7 +255,7 @@ class ExecutionService:
         with self._inflight_lock:
             self._inflight_lock.notify_all()
 
-    def pause_intake(self) -> None:
+    def pause_intake(self) -> bool:
         """Stop claiming, and wait for what is already claimed to finish.
 
         Called around a rebuild: items claimed against the old pipeline should
@@ -263,6 +263,9 @@ class ExecutionService:
         a claim still under way: the consumer holds its lock from the claim to
         the dispatch, so once that lock is free here the claim is counted, and
         every later pass finds the intake closed.
+
+        False when that took longer than `DRAIN_TIMEOUT_S`. The intake stays
+        closed either way; the caller decides whether to go on.
         """
         self._intake.clear()
         # So a consumer waiting for capacity sees the intake closed.
@@ -270,7 +273,7 @@ class ExecutionService:
         deadline = time.monotonic() + DRAIN_TIMEOUT_S
         if not self._consumer_lock.acquire(timeout=DRAIN_TIMEOUT_S):
             logger.warning("Rebuild did not wait out a claim")
-            return
+            return False
         self._consumer_lock.release()
         with self._inflight_lock:
             while self._running:
@@ -279,8 +282,9 @@ class ExecutionService:
                     logger.warning(
                         "Rebuild did not wait out %d cascades", self._running.total()
                     )
-                    return
+                    return False
                 self._inflight_lock.wait(remaining)
+        return True
 
     def resume_intake(self) -> None:
         self._intake.set()

backend/tests/flow/test_engine_hold.py

@@ -5,13 +5,14 @@ service stops claiming and its timers once claimed work has finished. Both
 wait without a limit and can be aborted.
 """
 
+import asyncio
 import threading
 import time
 
 import pytest
 
 from fluksio.flow import executor
-from fluksio.flow.controller import FlowController, LoadedNode
+from fluksio.flow.controller import FlowController, LoadedNode, RebuildBusy
 from fluksio.flow.executor import ExecutionService
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.namespace import Rename
@@ -149,6 +150,29 @@ def test_pause_intake_waits_out_a_claim_already_under_way(monkeypatch):
         service.stop()
 
 
+def test_a_reload_that_cannot_drain_leaves_the_pipeline_as_it_was(
+    tmp_path, monkeypatch
+):
+    monkeypatch.setattr(executor, "DRAIN_TIMEOUT_S", 0.05)
+    seen: list = []
+    running, finish, consume = _blocking(seen)
+    pipeline, service, source = _engine(consume)
+    controller = FlowController(FlowStore(tmp_path / "flows"), execution=service)
+    controller.pipeline = pipeline
+    service.start()
+    try:
+        source.inject({"reading": 1.0})
+        assert running.wait(5)
+        with pytest.raises(RebuildBusy):
+            asyncio.run(controller.reload())
+        assert controller.pipeline is pipeline
+        assert service._pipeline is pipeline
+        assert service._intake.is_set()
+    finally:
+        finish.set()
+        service.stop()
+
+
 def test_a_hold_waits_out_a_claim_already_under_way():
     seen: list = []
     queue = SlowClaim()