test(hydroflow): add tests for scheduler spinning #961 (#1171)

failing

hydroflow/tests/snapshots/surface_scheduling__nospin_issue_961@graphvis_dot.snap

@@ -0,0 +1,49 @@
+---
+source: hydroflow/tests/surface_scheduling.rs
+expression: "df.meta_graph().unwrap().to_dot(&Default::default())"
+---
+digraph {
+    node [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace", style=filled];
+    edge [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace"];
+    n1v1 [label="(n1v1) source_iter([1])", shape=invhouse, fillcolor="#88aaff"]
+    n2v1 [label="(n2v1) next_stratum()", shape=invhouse, fillcolor="#88aaff"]
+    n3v1 [label="(n3v1) persist()", shape=invhouse, fillcolor="#88aaff"]
+    n4v1 [label="(n4v1) defer_tick_lazy()", shape=invhouse, fillcolor="#88aaff"]
+    n5v1 [label="(n5v1) null()", shape=house, fillcolor="#ffff88"]
+    n6v1 [label="(n6v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n7v1 [label="(n7v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n8v1 [label="(n8v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n4v1 -> n5v1
+    n3v1 -> n6v1
+    n2v1 -> n7v1
+    n1v1 -> n8v1
+    n6v1 -> n4v1 [color=red]
+    n7v1 -> n3v1
+    n8v1 -> n2v1 [color=red]
+    subgraph "cluster n1v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_1v1\nstratum 0"
+        n1v1
+    }
+    subgraph "cluster n2v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_2v1\nstratum 1"
+        n2v1
+    }
+    subgraph "cluster n3v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_3v1\nstratum 1"
+        n3v1
+    }
+    subgraph "cluster n4v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_4v1\nstratum 0"
+        n4v1
+        n5v1
+    }
+}
+

hydroflow/tests/snapshots/surface_scheduling__nospin_issue_961@graphvis_mermaid.snap

@@ -0,0 +1,39 @@
+---
+source: hydroflow/tests/surface_scheduling.rs
+expression: "df.meta_graph().unwrap().to_mermaid(&Default::default())"
+---
+%%{init:{'theme':'base','themeVariables':{'clusterBkg':'#ddd','clusterBorder':'#888'}}}%%
+flowchart TD
+classDef pullClass fill:#8af,stroke:#000,text-align:left,white-space:pre
+classDef pushClass fill:#ff8,stroke:#000,text-align:left,white-space:pre
+classDef otherClass fill:#fdc,stroke:#000,text-align:left,white-space:pre
+linkStyle default stroke:#aaa
+1v1[\"(1v1) <code>source_iter([1])</code>"/]:::pullClass
+2v1[\"(2v1) <code>next_stratum()</code>"/]:::pullClass
+3v1[\"(3v1) <code>persist()</code>"/]:::pullClass
+4v1[\"(4v1) <code>defer_tick_lazy()</code>"/]:::pullClass
+5v1[/"(5v1) <code>null()</code>"\]:::pushClass
+6v1["(6v1) <code>handoff</code>"]:::otherClass
+7v1["(7v1) <code>handoff</code>"]:::otherClass
+8v1["(8v1) <code>handoff</code>"]:::otherClass
+4v1-->5v1
+3v1-->6v1
+2v1-->7v1
+1v1-->8v1
+6v1--o4v1; linkStyle 4 stroke:red
+7v1-->3v1
+8v1--x2v1; linkStyle 6 stroke:red
+subgraph sg_1v1 ["sg_1v1 stratum 0"]
+    1v1
+end
+subgraph sg_2v1 ["sg_2v1 stratum 1"]
+    2v1
+end
+subgraph sg_3v1 ["sg_3v1 stratum 1"]
+    3v1
+end
+subgraph sg_4v1 ["sg_4v1 stratum 0"]
+    4v1
+    5v1
+end
+

hydroflow/tests/snapshots/surface_scheduling__nospin_issue_961_complicated@graphvis_dot.snap

@@ -0,0 +1,72 @@
+---
+source: hydroflow/tests/surface_scheduling.rs
+expression: "df.meta_graph().unwrap().to_dot(&Default::default())"
+---
+digraph {
+    node [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace", style=filled];
+    edge [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace"];
+    n1v1 [label="(n1v1) source_iter([1])", shape=invhouse, fillcolor="#88aaff"]
+    n2v1 [label="(n2v1) union()", shape=invhouse, fillcolor="#88aaff"]
+    n3v1 [label="(n3v1) persist()", shape=invhouse, fillcolor="#88aaff"]
+    n4v1 [label="(n4v1) fold(|| 0, |accum, x| *accum += x)", shape=invhouse, fillcolor="#88aaff"]
+    n5v1 [label="(n5v1) defer_tick_lazy()", shape=invhouse, fillcolor="#88aaff"]
+    n6v1 [label="(n6v1) filter(|_| false)", shape=invhouse, fillcolor="#88aaff"]
+    n7v1 [label="(n7v1) tee()", shape=house, fillcolor="#ffff88"]
+    n8v1 [label="(n8v1) null()", shape=house, fillcolor="#ffff88"]
+    n9v1 [label="(n9v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n10v1 [label="(n10v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n11v1 [label="(n11v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+    n1v1 -> n2v1
+    n6v1 -> n7v1
+    n5v1 -> n6v1
+    n4v1 -> n9v1
+    n3v1 -> n10v1
+    n2v1 -> n3v1
+    n7v1 -> n8v1
+    n7v1 -> n11v1
+    n9v1 -> n5v1 [color=red]
+    n10v1 -> n4v1 [color=red]
+    n11v1 -> n2v1
+    subgraph "cluster n1v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_1v1\nstratum 0"
+        n1v1
+        n2v1
+        n3v1
+        subgraph "cluster_sg_1v1_var_double" {
+            label="var double"
+            n3v1
+        }
+        subgraph "cluster_sg_1v1_var_items" {
+            label="var items"
+            n2v1
+        }
+    }
+    subgraph "cluster n2v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_2v1\nstratum 1"
+        n4v1
+        subgraph "cluster_sg_2v1_var_double" {
+            label="var double"
+            n4v1
+        }
+    }
+    subgraph "cluster n3v1" {
+        fillcolor="#dddddd"
+        style=filled
+        label = "sg_3v1\nstratum 0"
+        n5v1
+        n6v1
+        n7v1
+        n8v1
+        subgraph "cluster_sg_3v1_var_double" {
+            label="var double"
+            n5v1
+            n6v1
+            n7v1
+        }
+    }
+}
+

hydroflow/tests/snapshots/surface_scheduling__nospin_issue_961_complicated@graphvis_mermaid.snap

@@ -0,0 +1,61 @@
+---
+source: hydroflow/tests/surface_scheduling.rs
+expression: "df.meta_graph().unwrap().to_mermaid(&Default::default())"
+---
+%%{init:{'theme':'base','themeVariables':{'clusterBkg':'#ddd','clusterBorder':'#888'}}}%%
+flowchart TD
+classDef pullClass fill:#8af,stroke:#000,text-align:left,white-space:pre
+classDef pushClass fill:#ff8,stroke:#000,text-align:left,white-space:pre
+classDef otherClass fill:#fdc,stroke:#000,text-align:left,white-space:pre
+linkStyle default stroke:#aaa
+1v1[\"(1v1) <code>source_iter([1])</code>"/]:::pullClass
+2v1[\"(2v1) <code>union()</code>"/]:::pullClass
+3v1[\"(3v1) <code>persist()</code>"/]:::pullClass
+4v1[\"(4v1) <code>fold(|| 0, |accum, x| *accum += x)</code>"/]:::pullClass
+5v1[\"(5v1) <code>defer_tick_lazy()</code>"/]:::pullClass
+6v1[\"(6v1) <code>filter(|_| false)</code>"/]:::pullClass
+7v1[/"(7v1) <code>tee()</code>"\]:::pushClass
+8v1[/"(8v1) <code>null()</code>"\]:::pushClass
+9v1["(9v1) <code>handoff</code>"]:::otherClass
+10v1["(10v1) <code>handoff</code>"]:::otherClass
+11v1["(11v1) <code>handoff</code>"]:::otherClass
+1v1-->2v1
+6v1-->7v1
+5v1-->6v1
+4v1-->9v1
+3v1-->10v1
+2v1-->3v1
+7v1-->8v1
+7v1-->11v1
+9v1--o5v1; linkStyle 8 stroke:red
+10v1--x4v1; linkStyle 9 stroke:red
+11v1-->2v1
+subgraph sg_1v1 ["sg_1v1 stratum 0"]
+    1v1
+    2v1
+    3v1
+    subgraph sg_1v1_var_double ["var <tt>double</tt>"]
+        3v1
+    end
+    subgraph sg_1v1_var_items ["var <tt>items</tt>"]
+        2v1
+    end
+end
+subgraph sg_2v1 ["sg_2v1 stratum 1"]
+    4v1
+    subgraph sg_2v1_var_double ["var <tt>double</tt>"]
+        4v1
+    end
+end
+subgraph sg_3v1 ["sg_3v1 stratum 0"]
+    5v1
+    6v1
+    7v1
+    8v1
+    subgraph sg_3v1_var_double ["var <tt>double</tt>"]
+        5v1
+        6v1
+        7v1
+    end
+end
+

hydroflow/tests/surface_scheduling.rs

@@ -1,7 +1,9 @@
 use std::error::Error;
+use std::time::Duration;
 
 use hydroflow::{assert_graphvis_snapshots, hydroflow_syntax, rassert_eq};
 use multiplatform_test::multiplatform_test;
+use tokio::time::timeout;
 
 #[multiplatform_test(test, wasm, env_tracing)]
 pub fn test_stratum_loop() {
@@ -79,7 +81,7 @@ async fn test_persist_stratum_run_async() -> Result<(), Box<dyn Error>> {
     };
     assert_graphvis_snapshots!(df);
 
-    tokio::time::timeout(std::time::Duration::from_millis(200), df.run_async())
+    timeout(Duration::from_millis(200), df.run_async())
         .await
         .expect_err("Expected time out");
 
@@ -121,3 +123,43 @@ pub fn test_issue_800_1050_fold_keyed() {
     assert_graphvis_snapshots!(df);
     df.run_available();
 }
+
+#[multiplatform_test(hydroflow, env_tracing)]
+async fn test_nospin_issue_961() {
+    let mut df = hydroflow_syntax! {
+        source_iter([1])
+            -> next_stratum()
+            -> persist()
+            -> defer_tick_lazy()
+            -> null();
+    };
+    assert_graphvis_snapshots!(df);
+
+    timeout(Duration::from_millis(100), df.run_available_async())
+        .await
+        .expect("Should not spin.");
+}
+
+#[multiplatform_test(hydroflow, env_tracing)]
+async fn test_nospin_issue_961_complicated() {
+    let mut df = hydroflow_syntax! {
+        source_iter([1]) -> items;
+        items = union();
+
+        double = items
+            -> persist()
+            -> fold(|| 0, |accum, x| *accum += x)
+            -> defer_tick_lazy()
+            -> filter(|_| false)
+            -> tee();
+
+        double -> null();
+
+        double -> items;
+    };
+    assert_graphvis_snapshots!(df);
+
+    timeout(Duration::from_millis(100), df.run_available_async())
+        .await
+        .expect("Should not spin.");
+}