Fix dora list listing twice a stopping dataflow when using multiple d…

…aemon. (#668) Previously, When a daemon stop it sends a message to the coordinator which will log it as an archived dataflow even though not every daemon has stopped within the dataflow. This PR should fix this issue.
dora-rs · Oct 8, 2024 · 3d6360d · 3d6360d
2 parents 8e4264e + 07ed728
commit 3d6360d
Showing 1 changed file with 12 additions and 6 deletions.
diff --git a/binaries/coordinator/src/lib.rs b/binaries/coordinator/src/lib.rs
@@ -269,16 +269,22 @@ async fn start_inner(
                 DataflowEvent::DataflowFinishedOnMachine { machine_id, result } => {
                     match running_dataflows.entry(uuid) {
                         std::collections::hash_map::Entry::Occupied(mut entry) => {
-                            // Archive finished dataflow
-                            archived_dataflows
-                                .entry(uuid)
-                                .or_insert_with(|| ArchivedDataflow::from(entry.get()));
-                            entry.get_mut().machines.remove(&machine_id);
+                            let dataflow = entry.get_mut();
+                            dataflow.machines.remove(&machine_id);
+                            tracing::info!(
+                                "removed machine id: {machine_id} from dataflow: {:#?}",
+                                dataflow.uuid
+                            );
                             dataflow_results
                                 .entry(uuid)
                                 .or_default()
                                 .insert(machine_id, result);
-                            if entry.get_mut().machines.is_empty() {
+
+                            if dataflow.machines.is_empty() {
+                                // Archive finished dataflow
+                                archived_dataflows
+                                    .entry(uuid)
+                                    .or_insert_with(|| ArchivedDataflow::from(entry.get()));
                                 let finished_dataflow = entry.remove();
                                 let reply = ControlRequestReply::DataflowStopped {
                                     uuid,