fix(dal,sdf): when finding change sets, ensure it belongs to the relevant workspace

lib/dal/src/change_set.rs

@@ -556,16 +556,18 @@ impl ChangeSet {
         Ok(())
     }
 
+    /// Finds a [`ChangeSet`] across all workspaces, ignoring the provided [`WorkspacePk`] on the
+    /// current [`DalContext`]
     #[instrument(
-        name = "change_set.find",
+        name = "change_set.find_across_workspaces",
         level = "debug",
         skip_all,
         fields(
             si.change_set.id = %change_set_id,
             si.workspace.id = Empty,
         ),
     )]
-    pub async fn find(
+    pub async fn find_across_workspaces(
         ctx: &DalContext,
         change_set_id: ChangeSetId,
     ) -> ChangeSetResult<Option<Self>> {
@@ -594,6 +596,52 @@ impl ChangeSet {
         }
     }
 
+    /// Find a change set within the [`WorkspacePk`] set for the current [`DalContext`]
+    #[instrument(
+        name = "change_set.find",
+        level = "debug",
+        skip_all,
+        fields(
+            si.change_set.id = %change_set_id,
+            si.workspace.id = Empty,
+        ),
+    )]
+    pub async fn find(
+        ctx: &DalContext,
+        change_set_id: ChangeSetId,
+    ) -> ChangeSetResult<Option<Self>> {
+        let span = current_span_for_instrument_at!("debug");
+        let workspace_id = ctx.workspace_pk()?;
+        let row = ctx
+            .txns()
+            .await?
+            .pg()
+            .query_opt(
+                "SELECT * FROM change_set_pointers WHERE id = $1 AND workspace_id = $2",
+                &[&change_set_id, &workspace_id],
+            )
+            .await?;
+
+        match row {
+            Some(row) => {
+                let change_set = Self::try_from(row)?;
+
+                if let Some(workspace_id) = change_set.workspace_id {
+                    span.record("si.workspace.id", workspace_id.to_string());
+                }
+                Ok(Some(change_set))
+            }
+            None => {
+                // warn here so we can see if something is requesting change sets cross workspace
+                warn!(
+                    si.workspace.id = %workspace_id,
+                    "Change Set Id: {change_set_id} not found for Workspace: {workspace_id}",
+                );
+                Ok(None)
+            }
+        }
+    }
+
     pub async fn list_active(ctx: &DalContext) -> ChangeSetResult<Vec<Self>> {
         let mut result = vec![];
         let rows = ctx

lib/dal/src/context.rs

@@ -403,9 +403,11 @@ impl DalContext {
         Ok(workspace)
     }
 
-    /// Update the context to use the most recent snapshot pointed to by the current `ChangeSetId`.
+    /// Update the context to use the most recent snapshot pointed to by the current [`ChangeSetId`].
+    /// Note: This does not guarantee that the [`ChangeSetId`] is contained within the [`WorkspacePk`]
+    /// for the current [`DalContext`]
     pub async fn update_snapshot_to_visibility(&mut self) -> TransactionsResult<()> {
-        let change_set = ChangeSet::find(self, self.change_set_id())
+        let change_set = ChangeSet::find_across_workspaces(self, self.change_set_id())
             .await
             .map_err(|err| TransactionsError::ChangeSet(err.to_string()))?
             .ok_or(TransactionsError::ChangeSetNotFound(self.change_set_id()))?;

lib/dal/src/workspace_snapshot/migrator.rs

@@ -87,7 +87,7 @@ impl SnapshotGraphMigrator {
         info!("Migrating {} snapshot(s)", open_change_sets.len(),);
 
         for change_set in open_change_sets {
-            let mut change_set = ChangeSet::find(ctx, change_set.id)
+            let mut change_set = ChangeSet::find_across_workspaces(ctx, change_set.id)
                 .await?
                 .ok_or(ChangeSetError::ChangeSetNotFound(change_set.id))?;
             if change_set.workspace_id.is_none() || change_set.status == ChangeSetStatus::Failed {

lib/dal/tests/integration_test/change_set.rs

@@ -3,7 +3,7 @@ use dal::{
     context::TransactionsErrorDiscriminants, DalContext, DalContextBuilder, HistoryActor,
     RequestContext, Workspace, WorkspacePk,
 };
-use dal::{ChangeSet, ChangeSetStatus, Component};
+use dal::{AccessBuilder, ChangeSet, ChangeSetStatus, Component};
 use dal_test::helpers::{
     create_component_for_default_schema_name_in_default_view, create_user, ChangeSetTestHelpers,
 };
@@ -255,6 +255,78 @@ async fn build_from_request_context_limits_to_change_sets_of_current_workspace(
         .is_err_and(|e| TransactionsErrorDiscriminants::BadWorkspaceAndChangeSet == e.into()));
 }
 
+#[test]
+async fn cannot_find_change_set_across_workspaces(
+    ctx: &mut DalContext,
+    ctx_builder: DalContextBuilder,
+) {
+    let user_1 = create_user(ctx).await.expect("Unable to create user");
+    let user_2 = create_user(ctx).await.expect("Unable to create user");
+    let user_1_workspace =
+        Workspace::new_from_builtin(ctx, WorkspacePk::generate(), "user_1 workspace", "token")
+            .await
+            .expect("Unable to create workspace");
+    let user_2_workspace =
+        Workspace::new_from_builtin(ctx, WorkspacePk::generate(), "user_2 workspace", "token")
+            .await
+            .expect("Unable to create workspace");
+    user_1
+        .associate_workspace(ctx, *user_1_workspace.pk())
+        .await
+        .expect("Unable to associate user with workspace");
+    user_2
+        .associate_workspace(ctx, *user_2_workspace.pk())
+        .await
+        .expect("Unable to associate user with workspace");
+    ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx)
+        .await
+        .expect("Unable to set up test data");
+
+    let request_context = RequestContext {
+        tenancy: dal::Tenancy::new(*user_2_workspace.pk()),
+        visibility: dal::Visibility {
+            change_set_id: user_2_workspace.default_change_set_id(),
+        },
+        history_actor: HistoryActor::User(user_2.pk()),
+        request_ulid: None,
+    };
+
+    let mut user_2_dal_ctx = ctx_builder
+        .build(request_context)
+        .await
+        .expect("built dal ctx for user 2");
+
+    //create a new change set for user 2
+    let user_2_change_set = ChangeSet::fork_head(&user_2_dal_ctx, "user 2")
+        .await
+        .expect("could not create change set");
+    ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(&mut user_2_dal_ctx)
+        .await
+        .expect("Unable to set up test data");
+
+    let user_1_tenancy = dal::Tenancy::new(*user_1_workspace.pk());
+    let access_builder = AccessBuilder::new(user_1_tenancy, HistoryActor::User(user_1.pk()), None);
+
+    let user_1_dal_context = ctx_builder
+        .build_head(access_builder)
+        .await
+        .expect("could not build dal context");
+
+    //first, let's ensure we can't find it when using the user 1 dal ctx
+    let user_2_change_set_unfound = ChangeSet::find(&user_1_dal_context, user_2_change_set.id)
+        .await
+        .expect("could not find change set");
+    assert!(user_2_change_set_unfound.is_none());
+
+    // But if we search for the change set across all workspaces, we find it
+    let user_2_change_set_found_harshly =
+        ChangeSet::find_across_workspaces(&user_1_dal_context, user_2_change_set.id)
+            .await
+            .expect("could not find change set");
+
+    assert!(user_2_change_set_found_harshly.is_some());
+}
+
 #[test]
 async fn build_from_request_context_allows_change_set_from_workspace_with_access(
     ctx: &mut DalContext,