Fix work recovery and reduce engine and chart overhead

backend/tests/flow/test_queue.py

@@ -242,6 +242,48 @@ def test_a_trigger_is_journaled_rather_than_run_on_the_spot():
     assert state["f.doubled"] == 6.0
 
 
+def test_a_state_failure_leaves_the_delivery_unacknowledged(monkeypatch):
+    pipeline, source, state, seen = _pipeline_with_a_consumer()
+    service = ExecutionService(pipeline._queue)
+    service.bind(pipeline)
+
+    def unavailable(*args, **kwargs):
+        raise ConnectionError("State unavailable")
+
+    monkeypatch.setattr(state, "record", unavailable)
+    source.inject({"reading": 3.0})
+    item = pipeline._queue.claim(1, 10)[0]
+    try:
+        service._dispatch(item)
+        service.pause_intake()
+        assert seen == []
+        assert pipeline._queue.stats()["pending"] == 1
+        assert service.inflight == 0
+    finally:
+        service.stop()
+
+
+def test_a_node_failure_still_acknowledges_its_delivery(monkeypatch):
+    pipeline, source, _state, seen = _pipeline_with_a_consumer()
+    service = ExecutionService(pipeline._queue)
+    service.bind(pipeline)
+
+    def broken(*args, **kwargs):
+        raise ValueError("Node failed")
+
+    monkeypatch.setattr(pipeline.get_node_by_id("f.consumer"), "f", broken)
+    source.inject({"reading": 3.0})
+    item = pipeline._queue.claim(1, 10)[0]
+    try:
+        service._dispatch(item)
+        service.pause_intake()
+        assert seen == []
+        assert pipeline._queue.stats()["pending"] == 0
+        assert service.inflight == 0
+    finally:
+        service.stop()
+
+
 def test_work_for_a_paused_flow_is_held_and_released_on_resume():
     pipeline, source, state, seen = _pipeline_with_a_consumer()
     service = ExecutionService(pipeline._queue)