Don't finish dataflow while there are still pending nodes

binaries/daemon/src/lib.rs

@@ -1377,10 +1377,11 @@ impl Daemon {
         if let Some(mut pid) = dataflow.running_nodes.remove(node_id).and_then(|n| n.pid) {
             pid.mark_as_stopped()
         }
-        if dataflow
-            .running_nodes
-            .iter()
-            .all(|(_id, n)| n.node_config.dynamic)
+        if !dataflow.pending_nodes.local_nodes_pending()
+            && dataflow
+                .running_nodes
+                .iter()
+                .all(|(_id, n)| n.node_config.dynamic)
         {
             let result = DataflowDaemonResult {
                 timestamp: self.clock.new_timestamp(),

binaries/daemon/src/pending.rs

@@ -59,6 +59,10 @@ impl PendingNodes {
         self.external_nodes = value;
     }
 
+    pub fn local_nodes_pending(&self) -> bool {
+        !self.local_nodes.is_empty()
+    }
+
     pub async fn handle_node_subscription(
         &mut self,
         node_id: NodeId,