fix: conditional forwarding in Async Dataplane (#2668)

Signed-off-by: Vigith Maurice <vigith@gmail.com>
Signed-off-by: Yashash H L <yashashhl25@gmail.com>
Co-authored-by: Yashash H L <yashashhl25@gmail.com>

Author: vigith

docs/user-guide/reference/conditional-forwarding.md

@@ -7,9 +7,31 @@ Below is a list of different logic operations that can be done on tags.
 - **or** - forwards the message if one of the tags specified is present in Message's tags.
 - **not** - forwards the message if all the tags specified are not present in Message's tags.
 
-For example, there's a UDF used to process numbers, and forward the result to different vertices based on the number is even or odd. In this case, you can set the `tag` to `even-tag` or `odd-tag` in each of the returned messages,
+For example, there's a UDF used to process numbers, and forward the result to different vertices based on the number is 
+even or odd. In this case, you can set the `tag` to `even-tag` or `odd-tag` in each of the returned messages,
 and define the edges as below:
 
+## Default Behavior
+
+* If no `conditions` are specified in the spec, the message will be forwarded to all the downstream vertices (independent
+  of the `tags` in the `Messages`).
+* In the code, if the `Messages` are not tagged but conditions are configured, we will still honour the edge conditions.
+
+## Syntax
+
+```yaml
+edges:
+  - from: ...
+    to: ...
+    conditions:
+      tags:
+        operator: ...
+        values:
+          - ...
+```
+
+## Example
+
 ```yaml
 edges:
   - from: p1

pkg/reconciler/validator/pipeline_validate.go

@@ -133,6 +133,14 @@ func ValidatePipeline(pl *dfv1.Pipeline) error {
 		} else {
 			toFromEdge[e.From+e.To] = true
 		}
+
+		if e.Conditions != nil {
+			if e.Conditions.Tags != nil {
+				if len(e.Conditions.Tags.Values) == 0 {
+					return fmt.Errorf("invalid edge: conditional forwarding requires at least one tag value")
+				}
+			}
+		}
 	}
 
 	if len(namesInEdges) != len(names) {

pkg/reconciler/validator/pipeline_validate_test.go

@@ -520,6 +520,17 @@ func TestValidatePipeline(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	t.Run("no tag values conditional forwarding", func(t *testing.T) {
+		testObj := testPipeline.DeepCopy()
+		operatorOr := dfv1.LogicOperatorOr
+		testObj.Spec.Edges[1].Conditions = &dfv1.ForwardConditions{Tags: &dfv1.TagConditions{
+			Operator: &operatorOr,
+			Values:   []string{}}}
+		err := ValidatePipeline(testObj)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid edge: conditional forwarding requires at least one tag value")
+	})
+
 	t.Run("allow conditional forwarding from source vertex or udf vertex", func(t *testing.T) {
 		testObj := testPipeline.DeepCopy()
 		operatorOr := dfv1.LogicOperatorOr

rust/numaflow-core/src/shared/forward.rs

@@ -4,21 +4,29 @@ use std::sync::Arc;
 use numaflow_models::models::ForwardConditions;
 
 /// Checks if the message should be written to downstream vertex based on the conditions
-/// and message tags. If no tags are provided but there are edge conditions present, we will
-/// still forward to all vertices.
+/// and message tags. If no tags are provided, we treat it as empty tags and still perform
+/// the condition check.
 pub(crate) fn should_forward(
     tags: Option<Arc<[String]>>,
     conditions: Option<Box<ForwardConditions>>,
 ) -> bool {
-    conditions.is_none_or(|conditions| {
-        conditions.tags.operator.as_ref().is_none_or(|operator| {
-            tags.as_ref().is_none_or(|tags| {
-                !conditions.tags.values.is_empty()
-                    && check_operator_condition(operator, &conditions.tags.values, tags)
-            })
-        })
-    })
+    // we should forward the message to downstream vertex if there are no edge conditions
+    let Some(conditions) = conditions else {
+        return true;
+    };
+
+    // Return true if there are no tags in the edge condition
+    if conditions.tags.values.is_empty() {
+        return true;
+    }
+
+    // Treat missing tags as empty and check the condition
+    let tags = tags.unwrap_or_else(|| Arc::from(vec![]));
+    // Default operator is "or", if not specified
+    let operator = conditions.tags.operator.as_deref().unwrap_or("or");
+    check_operator_condition(operator, &conditions.tags.values, &tags)
 }
+
 /// Determine the partition to write the message to by hashing the message id.
 pub(crate) fn determine_partition(
     message_id: String,
@@ -72,13 +80,6 @@ mod tests {
         assert!(result);
     }
 
-    #[tokio::test]
-    async fn test_evaluate_write_condition_no_tags() {
-        let conditions = ForwardConditions::new(TagConditions::new(vec!["tag1".to_string()]));
-        let result = should_forward(None, Some(Box::new(conditions)));
-        assert!(result);
-    }
-
     #[tokio::test]
     async fn test_evaluate_write_condition_and_operator() {
         let mut tag_conditions = TagConditions::new(vec!["tag1".to_string(), "tag2".to_string()]);
@@ -108,4 +109,61 @@ mod tests {
         let result = should_forward(tags, Some(Box::new(conditions)));
         assert!(result);
     }
+
+    #[tokio::test]
+    async fn test_empty_tags_with_and_operator() {
+        let mut tag_conditions = TagConditions::new(vec!["tag1".to_string(), "tag2".to_string()]);
+        tag_conditions.operator = Some("and".to_string());
+        let conditions = ForwardConditions::new(tag_conditions);
+
+        // Empty tags array (explicit empty)
+        let tags = Some(Arc::from(Vec::<String>::new()));
+        let result = should_forward(tags, Some(Box::new(conditions.clone())));
+        assert!(!result, "AND condition should fail with empty tags");
+
+        // None tags
+        let result = should_forward(None, Some(Box::new(conditions)));
+        assert!(!result, "AND condition should fail with None tags");
+    }
+
+    #[tokio::test]
+    async fn test_empty_tags_with_or_operator() {
+        let mut tag_conditions = TagConditions::new(vec!["tag1".to_string(), "tag2".to_string()]);
+        tag_conditions.operator = Some("or".to_string());
+        let conditions = ForwardConditions::new(tag_conditions);
+
+        // Empty tags array (explicit empty)
+        let tags = Some(Arc::from(Vec::<String>::new()));
+        let result = should_forward(tags, Some(Box::new(conditions.clone())));
+        assert!(!result, "OR condition should fail with empty tags");
+
+        // None tags
+        let result = should_forward(None, Some(Box::new(conditions)));
+        assert!(!result, "OR condition should fail with None tags");
+    }
+
+    #[tokio::test]
+    async fn test_empty_tags_with_not_operator() {
+        let mut tag_conditions = TagConditions::new(vec!["tag1".to_string(), "tag2".to_string()]);
+        tag_conditions.operator = Some("not".to_string());
+        let conditions = ForwardConditions::new(tag_conditions);
+
+        // Empty tags array (explicit empty)
+        let tags = Some(Arc::from(Vec::<String>::new()));
+        let result = should_forward(tags, Some(Box::new(conditions.clone())));
+        assert!(result, "NOT condition should pass with empty tags");
+
+        // None tags
+        let result = should_forward(None, Some(Box::new(conditions)));
+        assert!(result, "NOT condition should pass with None tags");
+    }
+
+    #[tokio::test]
+    async fn test_default_operator() {
+        let tag_conditions = TagConditions::new(vec!["tag1".to_string(), "tag2".to_string()]);
+        let conditions = ForwardConditions::new(tag_conditions);
+        let tags = Some(Arc::from(vec!["tag1".to_string(), "tag2".to_string()]));
+        let result = should_forward(tags, Some(Box::new(conditions)));
+        assert!(result);
+    }
 }