fix: prevent subscription cycles when adding subscriptions

Author: zacharyhamm

lib/dal/src/attribute/prototype.rs

@@ -571,6 +571,13 @@ impl AttributePrototype {
         ctx: &DalContext,
         prototype_id: AttributePrototypeId,
     ) -> AttributePrototypeResult<()> {
+        // Also remove all apas that use this prototype, to prevent hanging APAs
+        // during subscription creation
+        let apa_ids = AttributePrototypeArgument::list_ids_for_prototype(ctx, prototype_id).await?;
+        for apa_id in apa_ids {
+            AttributePrototypeArgument::remove(ctx, apa_id).await?;
+        }
+
         ctx.workspace_snapshot()?
             .remove_node_by_id(prototype_id)
             .await?;

lib/dal/src/attribute/value.rs

@@ -73,6 +73,7 @@ use crate::{
     TransactionsError,
     attribute::prototype::AttributePrototypeError,
     change_set::ChangeSetError,
+    component::subscription_graph::SubscriptionGraph,
     func::{
         FuncExecutionPk,
         argument::{
@@ -289,6 +290,11 @@ pub enum AttributeValueError {
         subscription: String,
         subscription_prop_kind: PropKind,
     },
+    #[error("Subscription would create a cycle between components")]
+    SubscriptionWouldCycle {
+        subscriber_av_id: AttributeValueId,
+        subscription: String,
+    },
     #[error("transactions error: {0}")]
     Transactions(#[from] TransactionsError),
     #[error("try lock error: {0}")]
@@ -1929,9 +1935,13 @@ impl AttributeValue {
 
     /// Set the source of this attribute value to one or more subscriptions.
     ///
-    /// This overwrites or overrides any existing value; if your intent is to append
-    /// subscriptions, you should first call AttributeValue::subscriptions() and append to that
-    /// list.
+    /// This overwrites or overrides any existing value; if your intent is to
+    /// append subscriptions, you should first call
+    /// AttributeValue::subscriptions() and append to that list.
+    ///
+    /// Note that subscriptions can create cycles in the graph. Use the
+    /// `SubscriptionGraph` to construct a dep graph of all components via their
+    /// subscriptions to check if a subscription creates a cycle.
     pub async fn set_to_subscription(
         ctx: &DalContext,
         subscriber_av_id: AttributeValueId,
@@ -1942,20 +1952,30 @@ impl AttributeValue {
         // Make sure the subscribed-to path is valid (i.e. it doesn't have to resolve
         // to a value *right now*, but it must be a valid path to the schema as it
         // exists--correct prop names, numeric indices for arrays, etc.)
+        let subscription_title = subscription.fmt_title(ctx).await;
         let subscription_prop_id = subscription.validate(ctx).await?;
         let subscription_prop_kind = Prop::kind(ctx, subscription_prop_id).await?;
         let subscriber_prop_kind = AttributeValue::prop_kind(ctx, subscriber_av_id).await?;
         if !subscription_prop_kind.js_compatible_with(subscriber_prop_kind) {
             return Err(AttributeValueError::SubscriptionTypeMismatch {
                 subscriber_av_id,
                 subscriber_prop_kind,
-                subscription: subscription.fmt_title(ctx).await,
+                subscription: subscription_title.clone(),
                 subscription_prop_kind,
             });
         }
 
         Self::set_to_subscription_unchecked(ctx, subscriber_av_id, subscription, func_id, reason)
-            .await
+            .await?;
+
+        if SubscriptionGraph::new(ctx).await?.is_cyclic() {
+            Err(AttributeValueError::SubscriptionWouldCycle {
+                subscriber_av_id,
+                subscription: subscription_title,
+            })
+        } else {
+            Ok(())
+        }
     }
 
     /// Set the source of this attribute value to one or more subscriptions.
@@ -1984,6 +2004,7 @@ impl AttributeValue {
         // Add the subscriptions as the argument
         let arg_id = FuncArgument::single_arg_for_func(ctx, func_id).await?;
         let apa = AttributePrototypeArgument::new(ctx, prototype_id, arg_id, subscription).await?;
+        dbg!("created apa for subscription", apa.id(), prototype_id);
         AttributePrototypeArgument::add_reason(ctx, apa.id(), reason).await?;
 
         // DVU all the way!

lib/dal/src/component.rs

@@ -188,6 +188,7 @@ pub mod properties;
 pub mod qualification;
 pub mod resource;
 pub mod socket;
+pub mod subscription_graph;
 pub mod suggestion;
 
 #[remain::sorted]

lib/dal/src/component/subscription_graph.rs

@@ -0,0 +1,49 @@
+//! Component subscription graph.
+//!
+//! Creates a dependency graph by walking value subscription edges, so that we
+//! can detect if the subscriptions would create a cycle
+//!
+use si_id::ComponentId;
+
+use super::{
+    Component,
+    ComponentResult,
+};
+use crate::{
+    AttributePrototype,
+    AttributeValue,
+    DalContext,
+    attribute::prototype::argument::AttributePrototypeArgument,
+    dependency_graph::DependencyGraph,
+};
+
+pub struct SubscriptionGraph {
+    inner: DependencyGraph<ComponentId>,
+}
+
+impl SubscriptionGraph {
+    pub async fn new(ctx: &DalContext) -> ComponentResult<Self> {
+        let components = Component::list_ids(ctx).await?;
+        let mut inner = DependencyGraph::new();
+
+        for component_id in components {
+            for (_, apa_id) in Component::subscribers(ctx, component_id).await? {
+                let prototype_id = AttributePrototypeArgument::prototype_id(ctx, apa_id).await?;
+                for av_id in AttributePrototype::attribute_value_ids(ctx, prototype_id).await? {
+                    let subscriber_component_id = AttributeValue::component_id(ctx, av_id).await?;
+                    // Don't add self-subscriptions
+                    if subscriber_component_id == component_id {
+                        continue;
+                    }
+                    inner.id_depends_on(subscriber_component_id, component_id);
+                }
+            }
+        }
+
+        Ok(Self { inner })
+    }
+
+    pub fn is_cyclic(&self) -> bool {
+        self.inner.is_cyclic()
+    }
+}

lib/dal/src/dependency_graph.rs

@@ -140,4 +140,8 @@ impl<T: Copy + std::cmp::Ord + std::cmp::Eq + std::cmp::PartialEq> DependencyGra
     pub fn all_ids(&self) -> impl Iterator<Item = T> {
         self.graph.node_weights().copied()
     }
+
+    pub fn is_cyclic(&self) -> bool {
+        petgraph::algo::is_cyclic_directed(&self.graph)
+    }
 }

lib/dal/tests/integration_test/attribute_value/subscription.rs

@@ -733,6 +733,47 @@ async fn remove_subscribed_component(ctx: &mut DalContext) -> Result<()> {
     Ok(())
 }
 
+#[test]
+async fn subscription_cycles(ctx: &mut DalContext) -> Result<()> {
+    create_testy_variant(ctx).await?;
+    component::create(ctx, "testy", "subscriber").await?;
+    component::create(ctx, "testy", "source").await?;
+    component::create(ctx, "testy", "source_2").await?;
+
+    value::subscribe(
+        ctx,
+        ("subscriber", "/domain/Value"),
+        ("source", "/domain/Value"),
+    )
+    .await?;
+
+    value::subscribe(
+        ctx,
+        ("source", "/domain/Value"),
+        ("source_2", "/domain/Value"),
+    )
+    .await?;
+
+    // Self subscription should be fine
+    value::subscribe(
+        ctx,
+        ("source_2", "/domain/Value3"),
+        ("source_2", "/domain/Value4"),
+    )
+    .await?;
+
+    let bad = value::subscribe(
+        ctx,
+        ("source_2", "/domain/Value2"),
+        ("subscriber", "/domain/Value2"),
+    )
+    .await;
+
+    assert!(bad.is_err(), "cyclic subscription should fail");
+
+    Ok(())
+}
+
 async fn create_testy_variant(ctx: &DalContext) -> Result<()> {
     // Make a variant with a Value prop
     variant::create(
@@ -744,6 +785,8 @@ async fn create_testy_variant(ctx: &DalContext) -> Result<()> {
                     props: [
                         { name: "Value", kind: "string" },
                         { name: "Value2", kind: "string" },
+                        { name: "Value3", kind: "string" },
+                        { name: "Value4", kind: "string" },
                         { name: "Values", kind: "array",
                             entry: { name: "ValuesItem", kind: "string" },
                         },

lib/dal/tests/integration_test/attribute_value/subscription/default_subscriptions.rs

@@ -126,10 +126,7 @@ async fn test_set_as_default_subscription_source(ctx: &DalContext) -> Result<()>
     assert_eq!(expected_defaults, possible_defaults);
 
     for default_sub in possible_defaults {
-        default_sub
-            .subscribe(ctx)
-            .await
-            .unwrap_or_else(|_| panic!("should be able to subscribe: {default_sub:?}"));
+        default_sub.subscribe(ctx).await?;
     }
 
     let conflicting_av_id = Component::attribute_values_for_prop_by_id(