From b32a7e49240080a96505c3c14a20672f31ffb0f1 Mon Sep 17 00:00:00 2001
From: haixuanTao <tao.xavier@outlook.com>
Date: Thu, 29 Feb 2024 14:30:56 +0100
Subject: [PATCH] Make `send_stdout_as` fail if there is more than one entry
 for a runtime node

---
 binaries/daemon/src/spawn.rs              |  4 +++-
 libraries/core/src/descriptor/mod.rs      | 12 ++++++------
 libraries/core/src/descriptor/validate.rs |  6 ++++++
 3 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/binaries/daemon/src/spawn.rs b/binaries/daemon/src/spawn.rs
index d64db99f..7a204bd4 100644
--- a/binaries/daemon/src/spawn.rs
+++ b/binaries/daemon/src/spawn.rs
@@ -58,7 +58,9 @@ pub async fn spawn_node(
         clock.clone(),
     )
     .await?;
-    let send_stdout_to = node.send_stdout_as();
+    let send_stdout_to = node
+        .send_stdout_as()
+        .context("Could not resolve `send_stdout_as` configuration")?;
 
     let mut child = match node.kind {
         dora_core::descriptor::CoreNodeKind::Custom(n) => {
diff --git a/libraries/core/src/descriptor/mod.rs b/libraries/core/src/descriptor/mod.rs
index 7f114e59..90ad74e6 100644
--- a/libraries/core/src/descriptor/mod.rs
+++ b/libraries/core/src/descriptor/mod.rs
@@ -1,7 +1,7 @@
 use crate::config::{
     CommunicationConfig, DataId, Input, InputMapping, NodeId, NodeRunConfig, OperatorId,
 };
-use eyre::{bail, Context, Result};
+use eyre::{bail, eyre, Context, Result};
 use serde::{Deserialize, Serialize};
 use serde_with_expand_env::with_expand_envs;
 use std::{
@@ -166,7 +166,7 @@ pub struct ResolvedNode {
 }
 
 impl ResolvedNode {
-    pub fn send_stdout_as(&self) -> Option<String> {
+    pub fn send_stdout_as(&self) -> Result<Option<String>> {
         match &self.kind {
             // TODO: Split stdout between operators
             CoreNodeKind::Runtime(n) => {
@@ -178,16 +178,16 @@ impl ResolvedNode {
                 if count == 1 && n.operators.len() > 1 {
                     warn!("All stdout from all operators of a runtime are going to be sent in the selected `send_stdout_as` operator.")
                 } else if count > 1 {
-                    warn!("More than one `send_stdout_as` operators for a runtime node. Selecting the first stdout operator.")
+                    return Err(eyre!("More than one `send_stdout_as` entries for a runtime node. Please only use one `send_stdout_as` per runtime."));
                 }
-                n.operators.iter().find_map(|op| {
+                Ok(n.operators.iter().find_map(|op| {
                     op.config
                         .send_stdout_as
                         .clone()
                         .map(|stdout| format!("{}/{}", op.id, stdout))
-                })
+                }))
             }
-            CoreNodeKind::Custom(n) => n.send_stdout_as.clone(),
+            CoreNodeKind::Custom(n) => Ok(n.send_stdout_as.clone()),
         }
     }
 }
diff --git a/libraries/core/src/descriptor/validate.rs b/libraries/core/src/descriptor/validate.rs
index 93063de9..316d3c58 100644
--- a/libraries/core/src/descriptor/validate.rs
+++ b/libraries/core/src/descriptor/validate.rs
@@ -93,6 +93,12 @@ pub fn check_dataflow(dataflow: &Descriptor, working_dir: &Path) -> eyre::Result
         };
     }
 
+    // Check that nodes can resolve `send_stdout_as`
+    for node in &nodes {
+        node.send_stdout_as()
+            .context("Could not resolve `send_stdout_as` configuration")?;
+    }
+
     if has_python_operator {
         check_python_runtime()?;
     }