fix: prevent subscription cycles when adding subscriptions

Author: zacharyhamm

lib/dal/src/attribute/attributes.rs

@@ -25,7 +25,10 @@ use super::{
         AttributePrototypeArgument,
         static_value::StaticArgumentValue,
     },
-    value::subscription::ValueSubscription,
+    value::{
+        AttributeValueError,
+        subscription::ValueSubscription,
+    },
 };
 use crate::{
     AttributePrototype,
@@ -36,7 +39,10 @@ use crate::{
     Prop,
     PropKind,
     WsEvent,
-    component::resource::ResourceData,
+    component::{
+        resource::ResourceData,
+        subscription_graph::SubscriptionGraph,
+    },
     func::intrinsics::IntrinsicFunc,
     prop::PropError,
     workspace_snapshot::node_weight::{
@@ -392,6 +398,13 @@ async fn update_attributes_inner(
                         )
                         .await?;
 
+                        if SubscriptionGraph::new(ctx).await?.is_cyclic() {
+                            return Err(AttributeValueError::SubscriptionWouldCycle {
+                                subscriber_av_id: target_av_id,
+                                subscription: subscription.fmt_title(ctx).await,
+                            })?;
+                        }
+
                         // Build the after value for the audit log!
                         after_value_source = match subscription.resolve(ctx).await? {
                             Some(resolved_av_id) => {

lib/dal/src/attribute/value.rs

@@ -289,6 +289,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 +1934,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,

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

@@ -3,6 +3,7 @@ use dal::{
     AttributeValue,
     Component,
     DalContext,
+    component::subscription_graph::SubscriptionGraph,
     func::authoring::FuncAuthoringClient,
 };
 use dal_test::{
@@ -733,6 +734,49 @@ 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?;
+
+    value::subscribe(
+        ctx,
+        ("source_2", "/domain/Value3"),
+        ("source_2", "/domain/Value4"),
+    )
+    .await?;
+
+    assert_eq!(false, SubscriptionGraph::new(ctx).await?.is_cyclic());
+
+    // Return to the first one, creating a cycle
+    value::subscribe(
+        ctx,
+        ("source_2", "/domain/Value2"),
+        ("subscriber", "/domain/Value2"),
+    )
+    .await?;
+
+    assert!(SubscriptionGraph::new(ctx).await?.is_cyclic());
+
+    Ok(())
+}
+
 async fn create_testy_variant(ctx: &DalContext) -> Result<()> {
     // Make a variant with a Value prop
     variant::create(
@@ -744,6 +788,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" },
                         },