check node input and output (#176)

* check node input and output * .
space-operator · Nov 23, 2024 · 900fd91 · 900fd91
1 parent dfc9466
commit 900fd91
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 14 deletions.
diff --git a/crates/flow-server/src/db_worker/user_worker.rs b/crates/flow-server/src/db_worker/user_worker.rs
@@ -490,15 +490,7 @@ impl ResponseError for StartError {
                 flow::Error::Canceled(_) => StatusCode::INTERNAL_SERVER_ERROR,
                 flow::Error::ValueNotFound(_) => StatusCode::INTERNAL_SERVER_ERROR,
                 flow::Error::CreateCmd(_) => StatusCode::INTERNAL_SERVER_ERROR,
-                flow::Error::BuildGraphError(e) => match e {
-                    flow::flow_graph::BuildGraphError::EdgesSameTarget => StatusCode::BAD_REQUEST,
-                    flow::flow_graph::BuildGraphError::EdgeSourceNotFound(_) => {
-                        StatusCode::BAD_REQUEST
-                    }
-                    flow::flow_graph::BuildGraphError::NodeNotFoundInPartialConfig(_) => {
-                        StatusCode::BAD_REQUEST
-                    }
-                },
+                flow::Error::BuildGraphError(_) => StatusCode::BAD_REQUEST,
                 flow::Error::GetFlow(e) => match e {
                     get_flow::Error::NotFound => StatusCode::NOT_FOUND,
                     get_flow::Error::Unauthorized => StatusCode::UNAUTHORIZED,

diff --git a/crates/flow/src/flow_graph.rs b/crates/flow/src/flow_graph.rs
@@ -360,6 +360,10 @@ pub enum BuildGraphError {
     EdgeSourceNotFound(NodeId),
     #[error("node not found in partial_config: {:?}", .0)]
     NodeNotFoundInPartialConfig(NodeId),
+    #[error("node {:?} has no input {:?}", .0.0, .0.1)]
+    NoInput((NodeId, String)),
+    #[error("node {:?} has no output {:?}", .0.0, .0.1)]
+    NoOutput((NodeId, String)),
 }
 
 impl FlowGraph {
@@ -455,7 +459,13 @@ impl FlowGraph {
                         continue;
                     }
                 }
-                Some(n) => (n.idx, n.command.input_is_required(&to.1).unwrap_or(true)),
+                Some(n) => {
+                    let required = n
+                        .command
+                        .input_is_required(&to.1)
+                        .ok_or_else(|| BuildGraphError::NoInput(to.clone()))?;
+                    (n.idx, required)
+                }
             };
             let (from_idx, optional) = match nodes.get(&from.0) {
                 None => {
@@ -480,10 +490,13 @@ impl FlowGraph {
                         continue;
                     }
                 }
-                Some(n) => (
-                    n.idx,
-                    n.command.output_is_optional(&from.1).unwrap_or(false),
-                ),
+                Some(n) => {
+                    let optional = n
+                        .command
+                        .output_is_optional(&from.1)
+                        .ok_or_else(|| BuildGraphError::NoOutput(from.clone()))?;
+                    (n.idx, optional)
+                }
             };
 
             g.add_edge(

diff --git a/lib/flow-lib/src/command/mod.rs b/lib/flow-lib/src/command/mod.rs
@@ -126,6 +126,11 @@ pub trait CommandTrait: Send + Sync + 'static {
         self.outputs()
             .into_iter()
             .find_map(|o| (o.name == name).then_some(o.optional))
+            .or_else(|| {
+                self.inputs()
+                    .into_iter()
+                    .find_map(|i| (i.name == name && i.passthrough).then_some(!i.required))
+            })
     }
 }