fix(dal): synchronize graph access in dependent values update

The concurrency types used by the WorkspaceSnapshot to provide interior
mutability in the tokio run time are *not* sufficient to prevent data
races when operating on the same graph on different threads, since our
graph operations are not "atomic" and the graph *WILL* end up being read
from different threads while a write operation is still in progress if
it is shared between threads for modification. For example after a node
is added but *before* the edges necessary to place that node in the
right spot in the graph.

A more general solution is needed to represent the concept of a
"transaction" when mutating the graph, to prevent other readers from
seeing incomplete graphs from in-progress write operations.

For now, we have added a lock to `DependentValuesUpdate` that ensures
the portion of the update that gathers the inputs to a function acquires
a read lock, (which is released as soon as the function is sent to the
executor, so that we don't hold on to it while executing in veritech),
while the portion of the update that updates the graph with the
execution result acquires a write lock.

Co-Authored-By: Jacob Helwig <jacob@systeminit.com>
Co-Authored-By: Nick Gerace <nick@systeminit.com>
Co-Authored-By: Scott Prutton <scott@systeminit.com>

Author: 4 people

lib/dal/src/attribute/value.rs

@@ -46,7 +46,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use telemetry::prelude::*;
 use thiserror::Error;
-use tokio::sync::TryLockError;
+use tokio::sync::{RwLock, TryLockError};
 use ulid::Ulid;
 
 pub use dependent_value_graph::DependentValueGraph;
@@ -497,7 +497,17 @@ impl AttributeValue {
     pub async fn execute_prototype_function(
         ctx: &DalContext,
         attribute_value_id: AttributeValueId,
+        read_lock: Arc<RwLock<()>>,
     ) -> AttributeValueResult<PrototypeExecutionResult> {
+        // When functions are being executed in the dependent values update job,
+        // we need to ensure we are not reading our input sources from a graph
+        // that is in the process of being mutated on another thread, since it
+        // will be incomplete (some nodes will not have all their edges added
+        // yet, for example, or a reference replacement may still be in
+        // progress). To handle this here, we grab a read lock, which will be
+        // locked for writing in the dependent values update job while the
+        // execution result is being written to the graph.
+        let read_guard = read_lock.read().await;
         let prototype_id = AttributeValue::prototype_id(ctx, attribute_value_id).await?;
         let prototype_func_id = AttributePrototype::func_id(ctx, prototype_id).await?;
         let destination_component_id =
@@ -619,6 +629,10 @@ impl AttributeValue {
             .await
             .map_err(|e| AttributeValueError::BeforeFunc(e.to_string()))?;
 
+        // We have gathered up all our inputs and so no longer need a lock on
+        // the graph. Be sure not to add graph walk operations below this drop
+        drop(read_guard);
+
         let (_, func_binding_return_value) = match FuncBinding::create_and_execute(
             ctx,
             prepared_func_binding_args.clone(),
@@ -721,8 +735,10 @@ impl AttributeValue {
         ctx: &DalContext,
         attribute_value_id: AttributeValueId,
     ) -> AttributeValueResult<()> {
+        // this lock is never locked for writing so is effectively a no-op here
+        let read_lock = Arc::new(RwLock::new(()));
         let execution_result =
-            AttributeValue::execute_prototype_function(ctx, attribute_value_id).await?;
+            AttributeValue::execute_prototype_function(ctx, attribute_value_id, read_lock).await?;
 
         AttributeValue::set_values_from_execution_result(ctx, attribute_value_id, execution_result)
             .await?;

lib/dal/src/job/definition/dependent_values_update.rs

@@ -1,13 +1,17 @@
 use std::{
     collections::{HashMap, HashSet},
     convert::TryFrom,
+    sync::Arc,
 };
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use telemetry::prelude::*;
 use thiserror::Error;
-use tokio::task::{JoinError, JoinSet};
+use tokio::{
+    sync::RwLock,
+    task::{JoinError, JoinSet},
+};
 use ulid::Ulid;
 
 //use crate::tasks::StatusReceiverClient;
@@ -58,6 +62,8 @@ pub struct DependentValuesUpdate {
     access_builder: AccessBuilder,
     visibility: Visibility,
     job: Option<JobInfo>,
+    #[serde(skip)]
+    set_value_lock: Arc<RwLock<()>>,
 }
 
 impl DependentValuesUpdate {
@@ -71,6 +77,7 @@ impl DependentValuesUpdate {
             access_builder,
             visibility,
             job: None,
+            set_value_lock: Arc::new(RwLock::new(())),
         })
     }
 }
@@ -149,6 +156,7 @@ impl DependentValuesUpdate {
                         id,
                         ctx.clone(),
                         attribute_value_id,
+                        self.set_value_lock.clone(),
                     ));
                     task_id_to_av_id.insert(id, attribute_value_id);
                     seen_ids.insert(attribute_value_id);
@@ -161,6 +169,12 @@ impl DependentValuesUpdate {
                 if let Some(finished_value_id) = task_id_to_av_id.remove(&task_id) {
                     match execution_result {
                         Ok(execution_values) => {
+                            // Lock the graph for writing inside this job. The
+                            // lock will be released when this guard is dropped
+                            // at the end of the scope.
+                            #[allow(unused_variables)]
+                            let write_guard = self.set_value_lock.write().await;
+
                             match AttributeValue::set_values_from_execution_result(
                                 ctx,
                                 finished_value_id,
@@ -216,10 +230,12 @@ async fn values_from_prototype_function_execution(
     task_id: Ulid,
     ctx: DalContext,
     attribute_value_id: AttributeValueId,
+    set_value_lock: Arc<RwLock<()>>,
 ) -> (Ulid, DependentValueUpdateResult<PrototypeExecutionResult>) {
-    let result = AttributeValue::execute_prototype_function(&ctx, attribute_value_id)
-        .await
-        .map_err(Into::into);
+    let result =
+        AttributeValue::execute_prototype_function(&ctx, attribute_value_id, set_value_lock)
+            .await
+            .map_err(Into::into);
 
     (task_id, result)
 }
@@ -234,6 +250,7 @@ impl TryFrom<JobInfo> for DependentValuesUpdate {
             access_builder: job.access_builder,
             visibility: job.visibility,
             job: Some(job),
+            set_value_lock: Arc::new(RwLock::new(())),
         })
     }
 }

lib/dal/src/workspace_snapshot.rs

@@ -101,6 +101,16 @@ pub enum WorkspaceSnapshotError {
 
 pub type WorkspaceSnapshotResult<T> = Result<T, WorkspaceSnapshotError>;
 
+/// The workspace graph. The concurrency types used here to give us interior
+/// mutability in the tokio run time are *not* sufficient to prevent data races
+/// when operating on the same graph on different threads, since our graph
+/// operations are not "atomic" and the graph *WILL* end up being read from
+/// different threads while a write operation is still in progress if it is
+/// shared between threads for modification. For example after a node is added
+/// but *before* the edges necessary to place that node in the right spot in the
+/// graph have been added. We need a more general solution here, but for now an
+/// example of synchronization when accessing a snapshot across threads can be
+/// found in [`crate::job::definition::DependentValuesUpdate`].
 #[derive(Debug, Clone)]
 pub struct WorkspaceSnapshot {
     address: Arc<RwLock<WorkspaceSnapshotAddress>>,