Repository: numaproj/numaflow

Issue:
Undocumented behaviour of conditional forwarding without tags
**Describe the bug**
The [Conditional forwarding docs](https://numaflow.numaproj.io/user-guide/reference/conditional-forwarding/) don't mention the fact that tagless messages are always forwarded which for me is a little unexpected and lost us a few hours while we tried to track down why we were getting messages in unexpected places.

https://github.com/numaproj/numaflow/blob/546561ab40f59d2e79d0ab2e9fee2402d594d9b1/rust/numaflow-core/src/shared/forward.rs#L6-L9



**To Reproduce**
Steps to reproduce the behavior:

1. Configure an edge with an and operator and a single tag
2. send a message with no tags
3. receive that message that should not have been forwarded

**Expected behavior**
For me the expected behaviour is that messages with no tags don't make it through `and` and `or` operators but at least noting this behaviour in the docs would help.


**Environment (please complete the following information):**

- Kubernetes: 1.32.3-gke.1785003
- Numaflow: v1.5.0-rc1


---


<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.

>For quick help and support, join our [slack](https://numaproj.slack.com/join/shared_invite/zt-19svuv47m-YKHhsQ~~KK9mBv1E7pNzfg#/shared-invite/email) channel.


Hints from issue comments:
@vigith: Thank you, @saty9. We will fix it in the code. The below will be the behavior after the fix. Please let us know if you have any concerns.

---

## 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.@KeranYang: > In the code, if the Messages are not tagged but conditions are configured, we will be dropping the messages.

@vigith What if the condition uses operator NOT?@yhl25: > > In the code, if the Messages are not tagged but conditions are configured, we will be dropping the messages.
> 
> [@vigith](https://github.com/vigith) What if the condition uses operator NOT?

fixed the comment, we will honour the edge conditions when tags are not specified in the message.@saty9: @vigith that sounds great thanks

Solution patch:
diff --git a/docs/user-guide/reference/conditional-forwarding.md b/docs/user-guide/reference/conditional-forwarding.md
index a08ae473..28706077 100644
--- a/docs/user-guide/reference/conditional-forwarding.md
+++ b/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
diff --git a/pkg/reconciler/validator/pipeline_validate.go b/pkg/reconciler/validator/pipeline_validate.go
index cb2d338e..603ef00a 100644
--- a/pkg/reconciler/validator/pipeline_validate.go
+++ b/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) {
diff --git a/rust/numaflow-core/src/shared/forward.rs b/rust/numaflow-core/src/shared/forward.rs
index 3b2cc226..2a98a679 100644
--- a/rust/numaflow-core/src/shared/forward.rs
+++ b/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);
+    }
 }



Test patch:
diff --git a/pkg/reconciler/validator/pipeline_validate_test.go b/pkg/reconciler/validator/pipeline_validate_test.go
index 35fb850b..039ecbab 100644
--- a/pkg/reconciler/validator/pipeline_validate_test.go
+++ b/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