Add comprehensive execution tracing (#177)

Signed-off-by: Michael X. Grey <mxgrey@intrinsic.ai>

Author: mxgrey

.github/workflows/ci_linux.yaml

@@ -25,6 +25,9 @@ jobs:
     steps:
     - uses: actions/checkout@v3
 
+    - name: Update apt index
+      run: sudo apt-get update
+
     - name: Get apt dependencies
       run: sudo apt-get install protobuf-compiler
 
@@ -41,7 +44,7 @@ jobs:
 
     - name: Test default features
       run: |
-        cargo test --workspace
+        cargo test
 
     - name: Test single_threaded_async
       run: |
@@ -51,12 +54,12 @@ jobs:
     - name: Test diagram
       run: |
         rm -rf target/debug/build
-        cargo test --workspace -F=diagram
+        cargo test -F=diagram
 
     - name: Test trace
       run: |
         rm -rf target/debug/build
-        cargo test --workspace -F=trace
+        cargo test -F=trace
 
     - name: Test all capabilities combined
       run: |

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "crossflow"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2024"
 authors = ["Grey <mxgrey@intrinsic.ai>"]
 license = "Apache-2.0"
@@ -64,7 +64,7 @@ zenoh-ext = "1.5.1"
 
 [dependencies]
 anyhow = { workspace = true }
-crossflow_derive = { path = "macros", version = "0.0.6" }
+crossflow_derive = { path = "macros", version = "0.0.7" }
 bevy_ecs = { workspace = true }
 bevy_utils = { workspace = true }
 bevy_derive = { workspace = true }

assets/figures/conditional-unreachability-cancel.svg

@@ -1,6 +1,6 @@
 [package]
 name = "crossflow_diagram_editor"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2024"
 authors = ["Teo Koon Peng <koonpeng@intrinsic.ai>"]
 license = "Apache-2.0"
@@ -29,7 +29,7 @@ path = "server/lib.rs"
 axum = { workspace = true, features = ["json"], default-features = false }
 bevy_app = { workspace = true }
 bevy_ecs = { workspace = true }
-crossflow = { version = "0.0.6", path = "..", features = ["diagram", "trace"] }
+crossflow = { version = "0.0.7", path = "..", features = ["diagram", "trace"] }
 clap = { workspace = true, features = ["derive"], optional = true }
 flate2 = { version = "1.1.1", optional = true }
 futures-util = { workspace = true }

diagram-editor/Cargo.toml

@@ -12,7 +12,7 @@ use axum::{
     routing::{self},
 };
 use bevy_ecs::{prelude::Entity, schedule::IntoScheduleConfigs};
-use crossflow::{Diagram, DiagramElementRegistry, OperationStarted, Outcome, RequestExt, trace};
+use crossflow::{Diagram, DiagramElementRegistry, Outcome, RequestExt, TracedEvent, trace};
 use serde::{Deserialize, Serialize};
 use std::{
     error::Error,
@@ -35,7 +35,7 @@ type WorkflowResponseResult =
     Result<(Outcome<serde_json::Value>, Entity), Box<dyn Error + Send + Sync>>;
 type WorkflowResponseSender = tokio::sync::oneshot::Sender<WorkflowResponseResult>;
 
-type WorkflowFeedback = OperationStarted;
+type WorkflowFeedback = TracedEvent;
 
 #[derive(bevy_ecs::component::Component)]
 struct FeedbackSender(tokio::sync::broadcast::Sender<WorkflowFeedback>);
@@ -320,26 +320,13 @@ fn execute_requests(
 }
 
 fn debug_feedback(
-    mut op_started: bevy_ecs::event::EventReader<trace::OperationStarted>,
-    feedback_query: bevy_ecs::system::Query<&FeedbackSender>,
+    mut op_started: bevy_ecs::event::EventReader<trace::TracedEvent>,
+    feedback_query: bevy_ecs::system::Query<(Entity, &FeedbackSender)>,
 ) {
     for ev in op_started.read() {
-        // not sure if it is working as intended, but the root session is in the 2nd last
-        // item, not the last as described in `session_stack` docs.
-        let session = match ev.session_stack.iter().rev().skip(1).next() {
-            Some(session) => session,
-            None => {
-                continue;
-            }
-        };
-        match feedback_query.get(*session) {
-            Ok(feedback_tx) => {
-                if let Err(e) = feedback_tx.0.send(ev.clone()) {
-                    error!("{}", e);
-                }
-            }
-            Err(_) => {
-                // the session has no feedback channel
+        for (session, channel) in &feedback_query {
+            if ev.event.is_for_session(session) {
+                let _ = channel.0.send(ev.clone());
             }
         }
     }

diagram-editor/server/api/executor.rs

@@ -1,6 +1,6 @@
 [package]
 name = "crossflow_diagram_editor_wasm"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2024"
 authors = ["Teo Koon Peng <koonpeng@intrinsic.ai>"]
 license = "Apache-2.0"
@@ -27,11 +27,11 @@ categories = [
 [dependencies]
 axum = { workspace = true, features = ["json"], default-features = false }
 bevy_app = { workspace = true }
-crossflow = { version = "0.0.6", path = "../..", features = [
+crossflow = { version = "0.0.7", path = "../..", features = [
   "diagram",
   "single_threaded_async",
 ] }
-crossflow_diagram_editor = { version = "0.0.6", path = "..", default-features = false }
+crossflow_diagram_editor = { version = "0.0.7", path = "..", default-features = false }
 mime_guess = { workspace = true }
 serde_json = { workspace = true }
 serde-wasm-bindgen = "0.6.5"

diagram-editor/wasm/Cargo.toml

@@ -105,21 +105,6 @@
         "buffers"
       ]
     },
-    "BufferIdentifier": {
-      "description": "Uniquely identify a buffer within a buffer map, either by name or by an\nindex value.",
-      "anyOf": [
-        {
-          "description": "Identify a buffer by name",
-          "type": "string"
-        },
-        {
-          "description": "Identify a buffer by an index value",
-          "type": "integer",
-          "format": "uint",
-          "minimum": 0
-        }
-      ]
-    },
     "BufferSchema": {
       "description": "Create a [`Buffer`][1] which can be used to store and pull data within\na scope.\n\nBy default the [`BufferSettings`][2] will keep the single last message\npushed to the buffer. You can change that with the optional `settings`\nproperty.\n\nUse the `\"serialize\": true` option to serialize the messages into\n[`JsonMessage`] before they are inserted into the buffer. This\nallows any serializable message type to be pushed into the buffer. If\nleft unspecified, the buffer will store the specific data type that gets\npushed into it. If the buffer inputs are not being serialized, then all\nincoming messages being pushed into the buffer must have the same type.\n\n[1]: crate::Buffer\n[2]: crate::BufferSettings\n\n# Examples\n```\n# crossflow::Diagram::from_json_str(r#\"\n{\n    \"version\": \"0.1.0\",\n    \"start\": \"fork_clone\",\n    \"ops\": {\n        \"fork_clone\": {\n            \"type\": \"fork_clone\",\n            \"next\": [\"num_output\", \"string_output\", \"all_num_buffer\", \"serialized_num_buffer\"]\n        },\n        \"num_output\": {\n            \"type\": \"node\",\n            \"builder\": \"num_output\",\n            \"next\": \"buffer_access\"\n        },\n        \"string_output\": {\n            \"type\": \"node\",\n            \"builder\": \"string_output\",\n            \"next\": \"string_buffer\"\n        },\n        \"string_buffer\": {\n            \"type\": \"buffer\",\n            \"settings\": {\n                \"retention\": { \"keep_last\": 10 }\n            }\n        },\n        \"all_num_buffer\": {\n            \"type\": \"buffer\",\n            \"settings\": {\n                \"retention\": \"keep_all\"\n            }\n        },\n        \"serialized_num_buffer\": {\n            \"type\": \"buffer\",\n            \"serialize\": true\n        },\n        \"buffer_access\": {\n            \"type\": \"buffer_access\",\n            \"buffers\": [\"string_buffer\"],\n            \"next\": \"with_buffer_access\"\n        },\n        \"with_buffer_access\": {\n            \"type\": \"node\",\n            \"builder\": \"with_buffer_access\",\n            \"next\": { \"builtin\": \"terminate\" }\n        }\n    }\n}\n# \"#)?;\n# Ok::<_, serde_json::Error>(())\n```",
       "type": "object",
@@ -461,6 +446,21 @@
         "err"
       ]
     },
+    "IdentifierRef": {
+      "description": "Uniquely identify something by a borrowed name or index.",
+      "anyOf": [
+        {
+          "description": "Identify by a name",
+          "type": "string"
+        },
+        {
+          "description": "Identify by an index value",
+          "type": "integer",
+          "format": "uint",
+          "minimum": 0
+        }
+      ]
+    },
     "InputExample": {
       "type": "object",
       "properties": {
@@ -505,7 +505,7 @@
           "description": "List of the keys in the `buffers` dictionary whose value should be cloned\ninstead of removed from the buffer (pulled) when the join occurs. Cloning\nthe value will leave the buffer unchanged after the join operation takes\nplace.",
           "type": "array",
           "items": {
-            "$ref": "#/$defs/BufferIdentifier"
+            "$ref": "#/$defs/IdentifierRef"
           }
         },
         "display_text": {

diagram.schema.json

@@ -1,11 +1,11 @@
 [package]
 name = "calculator"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2024"
 
 [dependencies]
-crossflow_diagram_editor = { path = "../../../diagram-editor", features = ["basic_executor"] }
-calculator_ops_catalog = { path = "../calculator_ops_catalog" }
+crossflow_diagram_editor = { path = "../../../diagram-editor", features = ["basic_executor"] , version = "0.0.7" }
+calculator_ops_catalog = { path = "../calculator_ops_catalog" , version = "0.0.7" }
 
 [dev-dependencies]
 assert_cmd = "2.1.2"

examples/diagram/calculator/Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "calculator_ops_catalog"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2024"
 description = "catalog of reusable calculator operations"
 
 [dependencies]
 bevy_app = { workspace = true }
-crossflow = { version = "0.0.6", path = "../../..", features = ["diagram"] }
+crossflow = { version = "0.0.7", path = "../../..", features = ["diagram"] }
 bevy_time = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

examples/diagram/calculator_ops_catalog/Cargo.toml

@@ -1,7 +1,7 @@
 use std::fmt::Write;
 
 use crossflow::{
-    AsyncMap, ConfigExample, DiagramElementRegistry, JsonMessage, NodeBuilderOptions, StreamPack,
+    Async, ConfigExample, DiagramElementRegistry, JsonMessage, NodeBuilderOptions, StreamPack,
 };
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -321,7 +321,7 @@ pub fn register(registry: &mut DiagramElementRegistry) {
             .with_config_examples(fibonacci_examples),
         |builder, config: Option<u64>| {
             builder.create_map(
-                move |input: AsyncMap<JsonMessage, FibonacciStream>| async move {
+                move |input: Async<JsonMessage, FibonacciStream>| async move {
                     let order = if let Some(order) = config {
                         order
                     } else if let JsonMessage::Number(number) = &input.request {