Repository: skypilot-org/skypilot
Pull request: [Jobs] Fix `is_chain` to check in- and out-degrees
<!-- Describe the changes in this PR -->

Resolve #4187 

- Adds `in_degree` checks to ensure the DAG has only one start and one end node.
- Ensures all nodes have at most one incoming and one outgoing edge.

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Code formatting: `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `conda deactivate; bash -i tests/backward_compatibility_tests.sh`


Problem statement:
Bug: `is_chain` function inaccurately detects chain structure
<!-- Describe the bug report / feature request here -->

The `is_chain` function fails to accurately detect chain structures. The current logic only checks that there's one node with an out-degree of 0 and all other nodes have an out-degree of 1. However, this overlooks cases where multiple nodes converge (i.e., nodes with an in-degree > 1), which are not valid for a chain structure.

**Example:**
Consider a graph with paths:
```
A → B → D
C → B
```
Here, `D` has an out-degree of 0, and all other nodes have an out-degree of 1, yet this is clearly not a single chain.

**Suggested Fix:**
To correctly identify a chain, the function should:
1. Verify there is exactly one start node (in-degree 0) and one end node (out-degree 0).
2. Ensure all other nodes have in-degree and out-degree of 1.

This correction will prevent false positives and ensure accurate chain detection.

Hints from issue discussion:
@cblmemo PTAL, thanks!

Commit message: fix: is_chain also checks out_degrees

Patch to review:
diff --git a/sky/dag.py b/sky/dag.py
index 4af5adc76b5..4d8f585d8f8 100644
--- a/sky/dag.py
+++ b/sky/dag.py
@@ -55,21 +55,21 @@ def get_graph(self):
         return self.graph
 
     def is_chain(self) -> bool:
-        # NOTE: this method assumes that the graph has no cycle.
-        is_chain = True
-        visited_zero_out_degree = False
-        for node in self.graph.nodes:
-            out_degree = self.graph.out_degree(node)
-            if out_degree > 1:
-                is_chain = False
-                break
-            elif out_degree == 0:
-                if visited_zero_out_degree:
-                    is_chain = False
-                    break
-                else:
-                    visited_zero_out_degree = True
-        return is_chain
+        """Check if the DAG is a linear chain of tasks."""
+
+        nodes = list(self.graph.nodes)
+        in_degrees = [self.graph.in_degree(node) for node in nodes]
+        out_degrees = [self.graph.out_degree(node) for node in nodes]
+
+        # Check out-degrees: all <=1 and exactly one node has out_degree ==0
+        out_degree_condition = (all(degree <= 1 for degree in out_degrees) and
+                                sum(degree == 0 for degree in out_degrees) == 1)
+
+        # Check in-degrees: all <=1 and exactly one node has in_degree ==0
+        in_degree_condition = (all(degree <= 1 for degree in in_degrees) and
+                               sum(degree == 0 for degree in in_degrees) == 1)
+
+        return out_degree_condition and in_degree_condition
 
 
 class _DagContext(threading.local):


Code review comments:
diff_hunk: @@ -55,21 +55,21 @@ def get_graph(self):
         return self.graph
 
     def is_chain(self) -> bool:
-        # NOTE: this method assumes that the graph has no cycle.
-        is_chain = True
-        visited_zero_out_degree = False
-        for node in self.graph.nodes:
-            out_degree = self.graph.out_degree(node)
-            if out_degree > 1:
-                is_chain = False
-                break
-            elif out_degree == 0:
-                if visited_zero_out_degree:
-                    is_chain = False
-                    break
-                else:
-                    visited_zero_out_degree = True
-        return is_chain
+        """Check if the DAG is a linear chain of tasks."""
+
+        nodes = list(self.graph.nodes)
+        in_degrees = [self.graph.in_degree(node) for node in nodes]
+        out_degrees = [self.graph.out_degree(node) for node in nodes]
+
+        # Check out-degrees: all <=1 and exactly one node has out_degree ==0
path: sky/dag.py
text: @user1:
```suggestion
        # Check out-degrees: all <= 1 and exactly one node has out_degree == 0
```

diff_hunk: @@ -55,21 +55,21 @@ def get_graph(self):
         return self.graph
 
     def is_chain(self) -> bool:
-        # NOTE: this method assumes that the graph has no cycle.
-        is_chain = True
-        visited_zero_out_degree = False
-        for node in self.graph.nodes:
-            out_degree = self.graph.out_degree(node)
-            if out_degree > 1:
-                is_chain = False
-                break
-            elif out_degree == 0:
-                if visited_zero_out_degree:
-                    is_chain = False
-                    break
-                else:
-                    visited_zero_out_degree = True
-        return is_chain
+        """Check if the DAG is a linear chain of tasks."""
+
+        nodes = list(self.graph.nodes)
+        in_degrees = [self.graph.in_degree(node) for node in nodes]
+        out_degrees = [self.graph.out_degree(node) for node in nodes]
+
+        # Check out-degrees: all <=1 and exactly one node has out_degree ==0
+        out_degree_condition = (all(degree <= 1 for degree in out_degrees) and
+                                sum(degree == 0 for degree in out_degrees) == 1)
+
+        # Check in-degrees: all <=1 and exactly one node has in_degree ==0
path: sky/dag.py
text: @user1:
```suggestion
        # Check in-degrees: all <= 1 and exactly one node has in_degree == 0
```