merge: #3986

3986: Frames fixes, features, and more precision when enqueuing DVU r=nickgerace a=britmyerss

This PR does the following: 

**Add precision to which values need to rerun when upserting/orphaning components by caching the state of inferred connections and only rerunning what has been deleted/added/changed**
- When changes are made to a component tree, we now correctly capture the initial, in-between (in the case of upsert), and final state of the tree in terms of all of the various inferred connections. 
- Based on the deltas of state as we make changes to the tree, we calculate which input sockets have new/different/changed inferred connections to output sockets and enqueue them in DVU (forcing them to rerun)
- For example, with a deep-ish nested tree, if you orphan a frame in the middle, the tree is split into two, and we need to walk both trees to find what needs to be rerun, as the orphaned frame might have had children that were taking inputs from an ancestor above the direct parent/child connection that was severed
- Got schooled/saved by `@nickgerace` on some rust

**Fix bug with nested propagation through up frames**
- We were incorrectly failing to proceed down to an up-frame's children if we're looking for which input sockets a given down frame is driving 
- For example, if a Down Frame -contains-> Up Frame -contains-> Component and the Component is taking a value from the Down Frame, when you change that value on the Down Frame, we were never finding the Component and you'd get left with stale data

**Enable up frames to take inputs from parent down frames**
- When discussing with `@keeb` - this is highly desired and comes up a lot as you rarely have an Up Frame without a Down Frame in its ancestry
- This enables, for example, a VPC Up Frame inside of a Region/Credential to infer the region/credential values, where previously we were not looking and forced the user to draw edges 
- This follows the same general intent as before - if there's ambiguity (for example, the VPC Up Frame is in a Region Down Frame, and also has a Region Component inside of it) - we don't infer any connection and force the user to resolve by drawing an edge

**Adds some telemetry that was useful while figuring this out** 
- connecting child/parent spans in Pinga/DVU so we see more of a connected waterfall in Honeycomb/Grafana
- Adding workspace and changeset ids to DVU jobs (to make it easier to query by when debugging specific issues)

**Tests!**
- Add a test for the new functionality for up frames
- Add a test to increase confidence that we're handling nested frames correctly on orphaning


<div><img src="https://media1.giphy.com/media/6iy6THGgvykxN0IA65/giphy.gif?cid=5a38a5a2yw96jdca3cfqjazt52av5xwbo94pzgt1f62bc65n&amp;ep=v1_gifs_search&amp;rid=giphy.gif&amp;ct=g" style="border:0;height:300px;width:300px"/><br/>via <a href="https://giphy.com/grantkolton/">grantkoltoons</a> on <a href="https://giphy.com/gifs/2d-get-back-up-again-i-knocked-down-6iy6THGgvykxN0IA65">GIPHY</a></div>

Co-authored-by: Brit Myers <[email protected]>

lib/dal-test/src/helpers.rs

@@ -4,8 +4,8 @@ use color_eyre::eyre::eyre;
 use color_eyre::Result;
 use dal::key_pair::KeyPairPk;
 use dal::{
-    AttributeValue, Component, ComponentId, DalContext, InputSocket, KeyPair, OutputSocket, Schema,
-    SchemaVariant, SchemaVariantId, User, UserPk,
+    AttributeValue, Component, ComponentId, ComponentType, DalContext, InputSocket, KeyPair,
+    OutputSocket, Schema, SchemaVariant, SchemaVariantId, User, UserPk,
 };
 use names::{Generator, Name};
 
@@ -70,6 +70,23 @@ pub async fn create_component_for_schema_name(
     Ok(Component::new(ctx, name.as_ref().to_string(), schema_variant_id).await?)
 }
 
+/// Creates a [`Component`] from the default [`SchemaVariant`] corresponding to a provided
+/// [`Schema`] name.
+pub async fn create_component_for_schema_name_with_type(
+    ctx: &DalContext,
+    schema_name: impl AsRef<str>,
+    name: impl AsRef<str>,
+    component_type: ComponentType,
+) -> Result<Component> {
+    let schema = Schema::find_by_name(ctx, schema_name)
+        .await?
+        .ok_or(eyre!("schema not found"))?;
+    let schema_variant_id = SchemaVariant::get_default_id_for_schema(ctx, schema.id()).await?;
+    let component = Component::new(ctx, name.as_ref().to_string(), schema_variant_id).await?;
+    component.set_type(ctx, component_type).await?;
+    Ok(component)
+}
+
 /// Creates a [`Component`] for a given [`SchemaVariantId`](SchemaVariant).
 pub async fn create_component_for_schema_variant(
     ctx: &DalContext,

lib/dal/src/component.rs

@@ -2332,7 +2332,7 @@ impl Component {
     /// if the input socket has arity many and the matches are all siblings
     ///
     /// Note: this does not check for whether data should actually flow between components
-    #[instrument(level = "debug", skip(ctx))]
+    #[instrument(level = "info", skip(ctx))]
     pub async fn find_available_inferred_connections_to_input_socket(
         ctx: &DalContext,
         input_socket_match: InputSocketMatch,
@@ -2362,22 +2362,42 @@ impl Component {
                     }
                 }
                 ComponentType::ConfigurationFrameUp => {
-                    // An up frame's input sockets are sourced from its children's output sockets
-                    // For now, we won't let down frames send outputs to parents and children
-                    // This might need to change, but we can change it when we've got a use case.
-                    let mut matches = Self::find_available_output_socket_match_in_descendants(
-                        ctx,
-                        input_socket_match,
-                        vec![
-                            ComponentType::ConfigurationFrameUp,
-                            ComponentType::Component,
-                        ],
-                    )
-                    .await?;
-                    // if there is more than one match, sort by component Ulid so they're
-                    // consistently ordered
-                    matches.sort_by_key(|output_socket| output_socket.component_id);
-                    matches
+                    // An up frame's input sockets are sourced from either its children's output sockets
+                    // or an ancestor.  Based on the input socket's arity, we match many (sorted by component ulid)
+                    // or if the arity is single, we return none
+                    let mut matches = vec![];
+                    let descendant_matches =
+                        Self::find_available_output_socket_match_in_descendants(
+                            ctx,
+                            input_socket_match,
+                            vec![
+                                ComponentType::ConfigurationFrameUp,
+                                ComponentType::Component,
+                            ],
+                        )
+                        .await?;
+                    matches.extend(descendant_matches);
+                    if let Some(ascendant_match) =
+                        Self::find_first_output_socket_match_in_ancestors(
+                            ctx,
+                            input_socket_match,
+                            vec![ComponentType::ConfigurationFrameDown],
+                        )
+                        .await?
+                    {
+                        matches.push(ascendant_match);
+                    }
+
+                    let input_socket =
+                        InputSocket::get_by_id(ctx, input_socket_match.input_socket_id).await?;
+                    if input_socket.arity() == SocketArity::One && matches.len() > 1 {
+                        vec![]
+                    } else {
+                        // if there is more than one match, sort by component Ulid so they're
+                        // consistently ordered
+                        matches.sort_by_key(|output_socket| output_socket.component_id);
+                        matches
+                    }
                 }
                 ComponentType::AggregationFrame => vec![],
             };
@@ -2425,9 +2445,10 @@ impl Component {
                         }
                     }
                 }
-                for child in Self::get_children_for_id(ctx, component_id).await? {
-                    work_queue.push_back(child);
-                }
+            }
+            // regardless whether the component type matches, we need to continue to descend
+            for child in Self::get_children_for_id(ctx, component_id).await? {
+                work_queue.push_back(child);
             }
         }
 
@@ -2876,7 +2897,7 @@ impl Component {
     /// Up Frame.
     ///
     /// Down Frames can drive Input Sockets of their children if the child is a Down Frame
-    /// or a Component.
+    /// or a Component or an Up Frame.
     #[instrument(level = "debug", skip(ctx))]
     pub async fn find_inferred_values_using_this_output_socket(
         ctx: &DalContext,
@@ -2908,21 +2929,21 @@ impl Component {
             }
             ComponentType::ConfigurationFrameDown => {
                 // if the type is a down frame, find all descendants
-                // who have a matching input socket AND are a Down Frame or Component
+                // who have a matching input socket AND are a Down Frame, Component, or Up Frame
                 Component::find_all_potential_inferred_input_socket_matches_in_descendants(
                     ctx,
                     output_socket_id,
                     component_id,
                     vec![
                         ComponentType::ConfigurationFrameDown,
                         ComponentType::Component,
+                        ComponentType::ConfigurationFrameUp,
                     ],
                 )
                 .await?
             }
-
             // we are not supporting aggregation frames right now
-            _ => vec![],
+            ComponentType::AggregationFrame => vec![],
         };
 
         Ok(maybe_target_sockets)