From bfd65bb104dfea686ba638d69af309ac9ed84f9e Mon Sep 17 00:00:00 2001 From: Zachary Hamm Date: Fri, 9 Jan 2026 10:39:46 -0600 Subject: [PATCH] feat(*): Apply to head happens in rebaser Currently we handle applies to head by calculating the set of updates in the current change set with respect to HEAD on the client side (e.g., in the rebaser or luminork) and then write a change set batch of those updates and send it to the rebaser. This is the wrong way to do it: with concurrent users, the set of updates calculated could easily be out of date by the time they reach the rebaser. Both HEAD and the change set we are updating could have moved forward, making our update set incomplete or incorrect. Further, we move the change set's status to applied on the client side as well, opening up further possibilities for races in a multiplayer setting. To fix this, we have moved the apply to head into the rebaser in a two phase commit approach: 1. First, on the request stream for the change set being applied, we send a request to begin applying the change set. This first locks all schemas and funcs that may have been in an editing state, writes the updated workspace snapshot, and then marks the change set with the "ApplyStarted" status. This prevents further updates from happening on the the change set since it is no longer "open". 2. Next, the *rebaser* and ONLY the rebaser makes requests an apply on the request stream for head. This calculates the update batch for an apply and performs the rebase, then moves the change set status to "Applied" if successful. If unsuccessful, the change set status will be reset to what it was when the prepare request was issued above, so that change sets are not left in "ApplyStarted" after an apply has failed. In step 1, the current workspace snapshot address of HEAD is sent. This is carried over to the Apply request in step 2. Before applying, the rebaser checks two things: First, that the workspace snaphot address for HEAD matches the one in the request. And then, it checks to see if the head snapshot has no DVU roots (if the DVU has completed). If either of these conditions fail, the rebaser will restart the apply process by sending a new BeginApply request (step 1). This loop will continue until the HEAD change set is at rest and the address has not changed since the begin apply request was sent. This ensures that concurrent applies to head don't lead to applies which are out of sync with head because they land before replays from *another* apply are able to run on the change set being applied. Further, this ensures that updates to the change set, or to head, cannot race an "apply", and that updates that land to a change set *after* an apply has started will be rejected (since they are no longer up to date). Also added here are request handlers for change set status changes and change set creation. The requests for these have not yet been integrated into the rebaser client, but they will similarly ensure that change set creation and status updates are handled in the correct order in multiplayer scenarios. --- lib/acceptable-macros/src/versioned.rs | 60 +- lib/dal-test/src/helpers/change_set.rs | 119 ++-- lib/dal/src/change_set.rs | 421 +++++++---- lib/dal/src/change_set/status.rs | 16 + lib/dal/src/context.rs | 59 +- lib/dal/src/workspace_snapshot/migrator.rs | 4 +- lib/dal/tests/integration_test/action.rs | 2 +- lib/dal/tests/integration_test/apply.rs | 86 +++ lib/dal/tests/integration_test/change_set.rs | 20 +- lib/dal/tests/integration_test/management.rs | 3 +- lib/dal/tests/integration_test/mod.rs | 1 + .../node_weight/schema_variant.rs | 4 +- .../integration_test/node_weight/view.rs | 63 +- .../src/change_set_processor_task.rs | 2 +- .../src/service/v1/change_sets/force_apply.rs | 4 +- .../v1/change_sets/request_approval.rs | 6 +- lib/rebaser-client/src/lib.rs | 307 ++++++-- .../src/api_types/enqueue_updates_request.rs | 55 +- .../api_types/enqueue_updates_request/v4.rs | 160 +++++ .../src/api_types/enqueue_updates_response.rs | 27 +- .../api_types/enqueue_updates_response/v2.rs | 90 +++ lib/rebaser-server/src/apply.rs | 265 +++++++ .../src/change_set_processor_task.rs | 662 ++++++++++++++++-- lib/rebaser-server/src/lib.rs | 1 + lib/rebaser-server/src/rebase.rs | 22 +- lib/sdf-core/src/dal_wrapper/change_set.rs | 11 +- lib/sdf-server/src/garbage_collection.rs | 2 +- .../src/service/v2/change_set/apply.rs | 2 +- .../v2/change_set/create_initialize_apply.rs | 7 +- .../src/service/v2/change_set/force_apply.rs | 8 +- lib/sdf-test/src/helpers.rs | 14 +- lib/si-events-rs/src/change_set_status.rs | 1 + 32 files changed, 2073 insertions(+), 431 deletions(-) create mode 100644 lib/dal/tests/integration_test/apply.rs create mode 100644 lib/rebaser-core/src/api_types/enqueue_updates_request/v4.rs create mode 100644 lib/rebaser-core/src/api_types/enqueue_updates_response/v2.rs create mode 100644 lib/rebaser-server/src/apply.rs diff --git a/lib/acceptable-macros/src/versioned.rs b/lib/acceptable-macros/src/versioned.rs index 455356b26c..3524a543d2 100644 --- a/lib/acceptable-macros/src/versioned.rs +++ b/lib/acceptable-macros/src/versioned.rs @@ -8,8 +8,10 @@ use proc_macro2::TokenStream as TokenStream2; use quote::quote; use syn::{ Data, + DataEnum, DataStruct, DeriveInput, + Fields, Ident, }; @@ -28,8 +30,9 @@ pub(crate) fn expand(input: TokenStream, errors: &mut manyhow::Emitter) -> Resul match data { Data::Struct(data_struct) => derive_from_struct(ident, data_struct, config, errors), - Data::Enum(_) | Data::Union(_) => { - bail!("current only struct types are supported for `#[derive(Versioned)]`") + Data::Enum(data_enum) => derive_from_enum(ident, data_enum, config, errors), + Data::Union(_) => { + bail!("union types are not supported for `#[derive(Versioned)]`") } } } @@ -64,3 +67,56 @@ fn derive_from_struct( } }) } + +fn derive_from_enum( + ident: Ident, + data: DataEnum, + config: Config, + _errors: &mut manyhow::Emitter, +) -> Result { + let version = config.version; + + // Verify that every variant has an `id` field + let mut match_arms = Vec::new(); + for variant in &data.variants { + let variant_ident = &variant.ident; + + let has_id_field = match &variant.fields { + Fields::Named(fields) => fields + .named + .iter() + .filter_map(|field| field.ident.as_ref()) + .any(|ident| ident == "id"), + Fields::Unnamed(_) => false, + Fields::Unit => false, + }; + + if !has_id_field { + bail!( + variant_ident, + "enum variant `{}` must have an `id` field for versioning", + variant_ident + ); + } + + match_arms.push(quote! { + Self::#variant_ident { id, .. } => (*id).into() + }); + } + + Ok(quote! { + impl acceptable::Versioned for #ident { + #[inline] + fn id(&self) -> acceptable::RequestId { + match self { + #(#match_arms),* + } + } + + #[inline] + fn message_version() -> u64 { + #version + } + } + }) +} diff --git a/lib/dal-test/src/helpers/change_set.rs b/lib/dal-test/src/helpers/change_set.rs index 04961f0bf1..9128828e0e 100644 --- a/lib/dal-test/src/helpers/change_set.rs +++ b/lib/dal-test/src/helpers/change_set.rs @@ -9,11 +9,9 @@ use color_eyre::{ use dal::{ ChangeSet, ChangeSetId, + ChangeSetStatus, ComponentId, DalContext, - Func, - Schema, - SchemaVariant, Ulid, action::{ Action, @@ -21,7 +19,10 @@ use dal::{ dependency_graph::ActionDependencyGraph, }, diagram::view::View, - workspace_snapshot::selector::WorkspaceSnapshotSelectorDiscriminants, + workspace_snapshot::{ + DependentValueRoot, + selector::WorkspaceSnapshotSelectorDiscriminants, + }, }; use si_db::{ ManagementFuncJobState, @@ -73,19 +74,17 @@ impl ChangeSetTestHelpers { pub async fn commit_and_update_snapshot_to_visibility(ctx: &mut DalContext) -> Result<()> { // The rebaser has responsibility for executing dvu jobs, so we should // prevent them from running in tests - let has_roots = ctx - .txns() + ctx.txns() .await? .job_queue() .clear_dependent_values_jobs() .await; ctx.blocking_commit().await?; + ctx.update_snapshot_to_visibility().await?; // But we have to wait until the dvu jobs complete - if has_roots { + if DependentValueRoot::roots_exist(ctx).await? { ChangeSet::wait_for_dvu(ctx, false).await?; - } else { - ctx.update_snapshot_to_visibility().await?; } Ok(()) @@ -122,13 +121,6 @@ impl ChangeSetTestHelpers { )) } - /// Apply Changeset To base Approvals - pub async fn apply_change_set_to_base_approvals(ctx: &mut DalContext) -> Result<()> { - ChangeSet::prepare_for_apply(ctx).await?; - Self::apply_change_set_to_base_approvals_without_prepare_step(ctx).await?; - Ok(()) - } - /// Switch to the [`ChangeSet`] corresponding to the provided ID. pub async fn switch_to_change_set( ctx: &mut DalContext, @@ -196,23 +188,17 @@ impl ChangeSetTestHelpers { /// Force Apply Changeset To base Approvals pub async fn force_apply_change_set_to_base_approvals(ctx: &mut DalContext) -> Result<()> { - ChangeSet::prepare_for_force_apply(ctx).await?; - Self::apply_change_set_to_base_approvals_without_prepare_step(ctx).await?; + ChangeSet::force_change_set_approval(ctx).await?; + Self::commit_and_update_snapshot_to_visibility(ctx).await?; + Self::apply_change_set_to_base_approvals_with_status_check(ctx).await?; Ok(()) } - /// Apply the change set to base for the approvals flow, but without performing the "prepare" - /// step. - /// - /// _Warning:_ if you do not know what to choose, use [Self::force_apply_change_set_to_base_approvals] - /// or [Self::apply_change_set_to_base_approvals] instead of this function. This function - /// should only be used if you have an alternative "prepare" workflow (e.g. testing fine grained - /// access control in sdf tests). - pub async fn apply_change_set_to_base_approvals_without_prepare_step( + /// Apply changeset to head, but only if Approved + pub async fn apply_change_set_to_base_approvals_with_status_check( ctx: &mut DalContext, ) -> Result<()> { - Self::commit_and_update_snapshot_to_visibility(ctx).await?; - Self::apply_change_set_to_base_inner(ctx).await?; + Self::apply_change_set_to_base_inner(ctx, false, true).await?; Ok(()) } @@ -233,8 +219,13 @@ impl ChangeSetTestHelpers { ) } - /// Applies the current change set to the base change set, waiting for replays to land on any open change sets. - pub async fn apply_change_set_to_base_inner(ctx: &mut DalContext) -> Result { + /// Applies the current change set to the base change set, waiting for + /// replays to land on any open change sets. + pub async fn apply_change_set_to_base_inner( + ctx: &mut DalContext, + skip_status_check: bool, + lock_schemas_and_funcs: bool, + ) -> Result { let mut open_change_sets = ChangeSet::list_active(ctx) .await? .iter() @@ -242,7 +233,39 @@ impl ChangeSetTestHelpers { .collect::>(); let had_updates = Self::has_updates(ctx).await?; - let applied_change_set = ChangeSet::apply_to_base_change_set(ctx).await?; + let applied_change_set = ChangeSet::get_by_id(ctx, ctx.change_set_id()).await?; + + if lock_schemas_and_funcs { + if skip_status_check { + ChangeSet::begin_apply_without_status_check(ctx).await? + } else { + ChangeSet::begin_apply(ctx).await? + } + } else { + // We only skip locking schemas/funcs in tests + if skip_status_check { + ChangeSet::begin_apply_no_lock_no_status_check(ctx).await? + } else { + ChangeSet::begin_apply_no_lock(ctx).await? + } + }; + + // Wait until the apply completes + let change_set_id = ctx.change_set_id(); + let mut change_set; + loop { + change_set = ChangeSet::get_by_id(ctx, change_set_id).await?; + if change_set.status != ChangeSetStatus::ApplyStarted { + break; + } + + tokio::time::sleep(Duration::from_millis(50)).await; + } + + if change_set.status != ChangeSetStatus::Applied { + color_eyre::eyre::bail!("change set apply failed"); + } + ctx.update_visibility_and_snapshot_to_visibility( applied_change_set.base_change_set_id.ok_or(eyre!( "base change set not found for change set: {}", @@ -257,7 +280,7 @@ impl ChangeSetTestHelpers { // waiting for the changes to reach the open change sets. if had_updates { let mut iters = 0; - // only do this for 10 seconds + // only do this for 30 seconds while !open_change_sets.is_empty() && iters < 1000 { let mut updated_sets = vec![]; for (change_set_id, original_updated_at) in &open_change_sets { @@ -274,7 +297,7 @@ impl ChangeSetTestHelpers { if open_change_sets.is_empty() { break; } - tokio::time::sleep(Duration::from_millis(10)).await; + tokio::time::sleep(Duration::from_millis(30)).await; iters += 1 } } @@ -282,35 +305,9 @@ impl ChangeSetTestHelpers { } /// Applies the current [`ChangeSet`] to its base [`ChangeSet`], leaving ctx pointed at HEAD. - /// 1. Prepares the apply by locking unlocked variants. - /// 2. Prepares the apply by Then, it updates the snapshot - /// to the visibility without using an editing [`ChangeSet`]. In other words, the resulting, - /// snapshot is "HEAD" without an editing [`ChangeSet`]. - /// - /// Also locks existing editing funcs and schema variants to mimic SDF pub async fn apply_change_set_to_base(ctx: &mut DalContext) -> Result { - // Lock all unlocked variants - for schema_id in Schema::list_ids(ctx).await? { - let schema = Schema::get_by_id(ctx, schema_id).await?; - let Some(variant) = SchemaVariant::get_unlocked_for_schema(ctx, schema_id).await? - else { - continue; - }; - - let variant_id = variant.id(); - - variant.lock(ctx).await?; - schema.set_default_variant_id(ctx, variant_id).await?; - } - // Lock all unlocked functions too - for func in Func::list_for_default_and_editing(ctx).await? { - if !func.is_locked { - func.lock(ctx).await?; - } - } - - ctx.commit().await?; - Self::apply_change_set_to_base_inner(ctx).await + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; + Self::apply_change_set_to_base_inner(ctx, true, true).await } /// Abandons the current [`ChangeSet`]. diff --git a/lib/dal/src/change_set.rs b/lib/dal/src/change_set.rs index dcd50e244e..b216a4d390 100644 --- a/lib/dal/src/change_set.rs +++ b/lib/dal/src/change_set.rs @@ -9,6 +9,18 @@ use chrono::{ DateTime, Utc, }; +use rebaser_client::{ + RebaserClient, + api_types::{ + enqueue_updates_request::BeginApplyMode, + enqueue_updates_response::{ + ApplyToHeadStatus, + BeginApplyStatus, + EnqueueUpdatesResponse, + EnqueueUpdatesResponseVCurrent, + }, + }, +}; use serde::{ Deserialize, Serialize, @@ -24,7 +36,6 @@ use si_db::{ change_set::FIND_ANCESTORS_QUERY, }; use si_events::{ - RebaseBatchAddressKind, WorkspaceSnapshotAddress, audit_log::AuditLogKind, merkle_tree_hash::MerkleTreeHash, @@ -34,11 +45,13 @@ use si_events::{ use si_id::{ ActionId, EntityId, + EventSessionId, UserPk, WorkspacePk, }; use si_layer_cache::LayerDbError; use telemetry::prelude::*; +use telemetry_utils::metric; use thiserror::Error; use tokio::time; @@ -64,12 +77,12 @@ use crate::{ self, BillingPublishError, }, + context::TransactionsResult, slow_rt::SlowRuntimeError, workspace_snapshot::{ DependentValueRoot, dependent_value_root::DependentValueRootError, graph::RebaseBatch, - selector::WorkspaceSnapshotSelectorDiscriminants, split_snapshot::{ SplitRebaseBatchVCurrent, SplitSnapshot, @@ -534,37 +547,84 @@ impl ChangeSet { /// First, transitions the status of the [`ChangeSet`] to [`ChangeSetStatus::NeedsApproval`] /// then [`ChangeSetStatus::Approved`]. Next, checks if DVU Roots still exist. Finally, /// lock every [`SchemaVariant`] and [`Func`] that is currently unlocked - pub async fn prepare_for_force_apply(ctx: &DalContext) -> ChangeSetResult<()> { + pub async fn force_change_set_approval(ctx: &DalContext) -> ChangeSetResult<()> { // first change the status to approved and who did it let mut change_set = ChangeSet::get_by_id(ctx, ctx.change_set_id()).await?; change_set.request_change_set_approval(ctx).await?; // then approve it change_set.approve_change_set_for_apply(ctx).await?; - // then do the rest - Self::prepare_for_apply(ctx).await + + Ok(()) } - /// First, checks if DVU Roots still exist. Next, ensures the [`ChangeSet`] has an - /// [`ChangeSetStatus::Approved`]. Finally, - /// lock every [`SchemaVariant`] and [`Func`] that is currently unlocked - pub async fn prepare_for_apply(ctx: &DalContext) -> ChangeSetResult<()> { - Self::prepare_for_apply_inner(ctx, false).await + /// First, checks if DVU Roots still exist. Next, ensures the [`ChangeSet`] + /// has an [`ChangeSetStatus::Approved`]. Then asks the begin applying the + /// change set. Returns the state of the change set prior to the prepare, so + /// that the rebaser can reset it if the apply fails. + pub async fn begin_apply(ctx: &DalContext) -> ChangeSetResult { + Self::begin_apply_inner(ctx, false, true).await } - /// This is a copy of [Self::prepare_for_apply], but skips the status check. This is because + /// This is a copy of [Self::begin_apply], but skips the status check. This is because /// sdf now handles the approvals flow as part of the fine grained access control work (i.e. /// SpiceDB is intentionally not accessible in the DAL). - pub async fn prepare_for_apply_without_status_check(ctx: &DalContext) -> ChangeSetResult<()> { - Self::prepare_for_apply_inner(ctx, true).await + pub async fn begin_apply_without_status_check( + ctx: &DalContext, + ) -> ChangeSetResult { + Self::begin_apply_inner(ctx, true, true).await + } + + pub async fn begin_apply_no_lock(ctx: &DalContext) -> ChangeSetResult { + Self::begin_apply_inner(ctx, false, false).await + } + + pub async fn begin_apply_no_lock_no_status_check( + ctx: &DalContext, + ) -> ChangeSetResult { + Self::begin_apply_inner(ctx, true, false).await + } + + #[instrument( + name = "change_set.lock_schemas_and_funcs_for_apply", + level = "info", + skip_all + )] + pub async fn lock_schemas_and_funcs_for_apply(ctx: &DalContext) -> ChangeSetResult<()> { + // Lock all unlocked variants + for schema_id in Schema::list_ids(ctx).await.map_err(Box::new)? { + let schema = Schema::get_by_id(ctx, schema_id).await.map_err(Box::new)?; + let Some(variant) = SchemaVariant::get_unlocked_for_schema(ctx, schema_id) + .await + .map_err(Box::new)? + else { + continue; + }; + + let variant_id = variant.id(); + + variant.lock(ctx).await.map_err(Box::new)?; + schema + .set_default_variant_id(ctx, variant_id) + .await + .map_err(Box::new)?; + } + // Lock all unlocked functions too + for func in Func::list_all(ctx).await.map_err(Box::new)? { + if !func.is_locked { + func.lock(ctx).await.map_err(Box::new)?; + } + } + Ok(()) } // TODO(nick): now that the fine grained access control flag is gone, we can collapse the two // outer methods and chase down tests continuing to use the old method. - async fn prepare_for_apply_inner( + async fn begin_apply_inner( ctx: &DalContext, dangerous_skip_status_check: bool, - ) -> ChangeSetResult<()> { + lock_schemas_and_funcs: bool, + ) -> ChangeSetResult { // Ensure that DVU roots are empty before continuing. if DependentValueRoot::roots_exist(ctx).await? { // TODO(nick): we should consider requiring this check in integration tests too. Why did I @@ -587,34 +647,28 @@ impl ChangeSet { } } - // Lock all unlocked variants - for schema_id in Schema::list_ids(ctx).await.map_err(Box::new)? { - let schema = Schema::get_by_id(ctx, schema_id).await.map_err(Box::new)?; - let Some(variant) = SchemaVariant::get_unlocked_for_schema(ctx, schema_id) - .await - .map_err(Box::new)? - else { - continue; - }; - - let variant_id = variant.id(); + let head_change_set_id = ctx.get_workspace_default_change_set_id().await?; + let head_change_set_address = ChangeSet::get_by_id(ctx, head_change_set_id) + .await? + .workspace_snapshot_address; + + let previous_change_set_status = begin_apply_with_reply( + ctx.rebaser(), + ctx.workspace_pk()?, + ctx.change_set_id(), + None, + head_change_set_id, + head_change_set_address, + ctx.event_session_id(), + if lock_schemas_and_funcs { + BeginApplyMode::LockSchemasAndFuncs + } else { + BeginApplyMode::NoLock + }, + ) + .await?; - variant.lock(ctx).await.map_err(Box::new)?; - schema - .set_default_variant_id(ctx, variant_id) - .await - .map_err(Box::new)?; - } - // Lock all unlocked functions too - for func in Func::list_for_default_and_editing(ctx) - .await - .map_err(Box::new)? - { - if !func.is_locked { - func.lock(ctx).await.map_err(Box::new)?; - } - } - Ok(()) + Ok(previous_change_set_status) } pub async fn approve_change_set_for_apply(&mut self, ctx: &DalContext) -> ChangeSetResult<()> { @@ -755,6 +809,33 @@ impl ChangeSet { } } + pub async fn list_replayable(ctx: &DalContext) -> ChangeSetResult> { + let mut result = vec![]; + let rows = ctx + .txns() + .await? + .pg() + .query( + "SELECT * from change_set_pointers WHERE workspace_id = $1 AND status IN ($2, $3, $4, $5, $6, $7)", + &[ + &ctx.tenancy().workspace_pk_opt(), + &ChangeSetStatus::Open.to_string(), + &ChangeSetStatus::NeedsApproval.to_string(), + &ChangeSetStatus::NeedsAbandonApproval.to_string(), + &ChangeSetStatus::Approved.to_string(), + &ChangeSetStatus::Rejected.to_string(), + &ChangeSetStatus::ApplyStarted.to_string(), + ], + ) + .await?; + + for row in rows { + result.push(Self::try_from(row)?); + } + + Ok(result) + } + pub async fn list_active(ctx: &DalContext) -> ChangeSetResult> { let mut result = vec![]; let rows = ctx @@ -884,31 +965,33 @@ impl ChangeSet { Ok(result) } - /// Applies the current [`ChangeSet`] in the provided [`DalContext`]. [`Actions`](Action) - /// are enqueued as needed and only done so if the base [`ChangeSet`] is "HEAD" (i.e. - /// the default [`ChangeSet`] of the [`Workspace`]). - /// Also sends the relevant WSEvent + /// This should *only* *EVER* be called by the rebaser after the begin apply step #[instrument(level = "info", skip_all)] - pub async fn apply_to_base_change_set(ctx: &mut DalContext) -> ChangeSetApplyResult { - let base_change_set_id = ctx.get_workspace_default_change_set_id().await?; - - if ctx.change_set_id() == base_change_set_id { - return Err(ChangeSetApplyError::CannotApplyToItself(base_change_set_id)); + pub async fn apply_to_base_change_set( + ctx: &mut DalContext, + previous_change_set_status: ChangeSetStatus, + head_change_set_address: WorkspaceSnapshotAddress, + mode: Option, + ) -> ChangeSetApplyResult<()> { + let head_change_set_id = ctx.get_workspace_default_change_set_id().await?; + + if ctx.change_set_id() == head_change_set_id { + return Err(ChangeSetApplyError::CannotApplyToItself(head_change_set_id)); } - // Apply to the base change with the current change set (non-editing) and commit. - let mut change_set_to_be_applied = Self::get_by_id(ctx, ctx.change_set_id()).await?; - - ctx.update_visibility_and_snapshot_to_visibility(ctx.change_set_id()) - .await?; - change_set_to_be_applied - .apply_to_base_change_set_inner(ctx) - .await?; - - // This is just to send the ws events - ctx.blocking_commit_no_rebase().await?; + apply_to_head_with_reply( + ctx.rebaser(), + ctx.workspace_pk()?, + head_change_set_id, + head_change_set_address, + ctx.change_set_id(), + ctx.event_session_id(), + previous_change_set_status, + mode, + ) + .await?; - Ok(change_set_to_be_applied) + Ok(()) } #[instrument( @@ -971,83 +1054,6 @@ impl ChangeSet { .map_err(Box::new)?) } - /// Applies the current [`ChangeSet`] in the provided [`DalContext`] to its base - /// [`ChangeSet`]. This involves performing a rebase request, updating the status - /// of the [`ChangeSet`] accordingly, and publishing a WSEvent - /// - /// This function neither changes the visibility nor the snapshot after performing the - /// aforementioned actions. - async fn apply_to_base_change_set_inner(&mut self, ctx: &DalContext) -> ChangeSetResult<()> { - let workspace_id = self - .workspace_id - .ok_or(ChangeSetError::NoWorkspacePkSet(self.id))?; - let base_change_set_id = self - .base_change_set_id - .ok_or(ChangeSetError::NoBaseChangeSet(self.id))?; - - let snapshot_kind: WorkspaceSnapshotSelectorDiscriminants = - ctx.workspace_snapshot().map_err(Box::new)?.into(); - - let maybe_rebase_batch_address = match snapshot_kind { - WorkspaceSnapshotSelectorDiscriminants::LegacySnapshot => { - if let Some(rebase_batch) = - self.detect_updates_that_will_be_applied_legacy(ctx).await? - { - Some(RebaseBatchAddressKind::Legacy( - ctx.write_legacy_rebase_batch(rebase_batch).await?, - )) - } else { - None - } - } - WorkspaceSnapshotSelectorDiscriminants::SplitSnapshot => { - if let Some(rebase_batch) = - self.detect_updates_that_will_be_applied_split(ctx).await? - { - Some(RebaseBatchAddressKind::Split( - ctx.write_split_snapshot_rebase_batch(rebase_batch).await?, - )) - } else { - None - } - } - }; - - if let Some(rebase_batch_address) = maybe_rebase_batch_address { - let (request_id, reply_fut) = ctx - .run_rebase_from_change_set_with_reply( - workspace_id, - base_change_set_id, - rebase_batch_address, - self.id, - ) - .await?; - - let reply_fut = reply_fut.instrument(info_span!( - "rebaser_client.await_response", - si.workspace.id = %workspace_id, - si.change_set.id = %base_change_set_id, - )); - - // Wait on response from Rebaser after request has processed - let timeout = Duration::from_secs(60); - let _reply = time::timeout(timeout, reply_fut) - .await - .map_err(|_elapsed| { - TransactionsError::RebaserReplyDeadlineElasped(timeout, request_id) - })??; - } - - self.update_status(ctx, ChangeSetStatus::Applied).await?; - let user = Self::extract_userid_from_context(ctx).await; - WsEvent::change_set_applied(ctx, self.id, base_change_set_id, user) - .await? - .publish_on_commit(ctx) - .await?; - - Ok(()) - } - /// Returns a new [`ChangeSetId`](ChangeSet) if a new [`ChangeSet`] was created. /// Also writes an audit log event to head (so we don't have to handle this in every route handler) pub async fn force_new(ctx: &mut DalContext) -> ChangeSetResult> { @@ -1243,3 +1249,138 @@ pub async fn calculate_checksum( } Ok(hasher.finalize()) } + +#[instrument( + level="info", + skip_all, + fields( + si.change_set.id = %change_set_id, + si.workspace.id = %workspace_pk, + si.head_change_set.id = %head_change_set_id, + ), + )] +#[inline] +#[allow(clippy::too_many_arguments)] +async fn begin_apply_with_reply( + rebaser: &RebaserClient, + workspace_pk: WorkspacePk, + change_set_id: ChangeSetId, + previous_change_set_status: Option, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + event_session_id: EventSessionId, + mode: BeginApplyMode, +) -> TransactionsResult { + let timeout = Duration::from_secs(60); + let metric_label = format!("{workspace_pk}:{change_set_id}"); + metric!(counter.dal.begin_apply_request = 1, label = metric_label); + let (request_id, reply_fut) = rebaser + .enqueue_begin_apply_with_retry( + workspace_pk, + change_set_id, + previous_change_set_status.map(Into::into), + head_change_set_id, + head_change_set_address, + event_session_id, + mode, + ) + .await?; + + let reply_fut = reply_fut.instrument(info_span!( + "rebaser_client.await_response", + si.workspace.id = %workspace_pk, + si.change_set.id = %change_set_id, + )); + + // Wait on response from Rebaser after request has processed + let reply = time::timeout(timeout, reply_fut) + .await + .map_err(|_elapsed| { + TransactionsError::RebaserReplyDeadlineElasped(timeout, request_id) + })??; + + metric!(counter.dal.begin_apply_requested = -1, label = metric_label); + + let EnqueueUpdatesResponse::V2(EnqueueUpdatesResponseVCurrent::BeginApplyToHead { + status, .. + }) = reply + else { + return Err(TransactionsError::RebaserUnexpectedResponse(reply.clone())); + }; + + match status { + BeginApplyStatus::Success { + previous_change_set_status, + } => Ok(previous_change_set_status.into()), + BeginApplyStatus::Error { message } => { + Err(TransactionsError::BeginApplyFailed(change_set_id, message)) + } + } +} + +#[instrument( + level="info", + skip_all, + fields( + si.change_set.id = %head_change_set_id, + si.workspace.id = %workspace_pk, + si.change_set_to_apply.id = %change_set_to_apply_id, + ), + )] +#[inline] +#[allow(clippy::too_many_arguments)] +async fn apply_to_head_with_reply( + rebaser: &RebaserClient, + workspace_pk: WorkspacePk, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + change_set_to_apply_id: ChangeSetId, + event_session_id: EventSessionId, + previous_change_set_status: ChangeSetStatus, + mode: Option, +) -> TransactionsResult<()> { + let timeout = Duration::from_secs(60); + let metric_label = format!("{workspace_pk}:{head_change_set_id}"); + metric!(counter.dal.apply_requested = 1, label = metric_label); + let (request_id, reply_fut) = rebaser + .enqueue_apply_to_head_with_reply( + workspace_pk, + head_change_set_id, + head_change_set_address, + change_set_to_apply_id, + event_session_id, + previous_change_set_status.into(), + mode, + ) + .await?; + + let reply_fut = reply_fut.instrument(info_span!( + "rebaser_client.await_response", + si.workspace.id = %workspace_pk, + si.change_set.id = %head_change_set_id, + )); + + // Wait on response from Rebaser after request has processed + let reply = time::timeout(timeout, reply_fut) + .await + .map_err(|_elapsed| { + TransactionsError::RebaserReplyDeadlineElasped(timeout, request_id) + })??; + + metric!(counter.dal.apply_requested = -1, label = metric_label); + + let EnqueueUpdatesResponse::V2(EnqueueUpdatesResponseVCurrent::ApplyToHead { status, .. }) = + reply + else { + return Err(TransactionsError::RebaserUnexpectedResponse(reply.clone())); + }; + + match status { + ApplyToHeadStatus::Success | ApplyToHeadStatus::Retrying => Ok(()), + ApplyToHeadStatus::Error { message } => Err(TransactionsError::ApplyFailed( + head_change_set_id, + change_set_to_apply_id, + message, + )), + } +} diff --git a/lib/dal/src/change_set/status.rs b/lib/dal/src/change_set/status.rs index e9c728de18..3f77403c46 100644 --- a/lib/dal/src/change_set/status.rs +++ b/lib/dal/src/change_set/status.rs @@ -32,6 +32,8 @@ pub enum ChangeSetStatus { Abandoned, /// Applied this changeset to its parent Applied, + /// An apply has begun, but not yet finished + ApplyStarted, /// Approved by relevant parties and ready to be applied Approved, /// Migration of Workspace Snapshot for this change set failed @@ -48,6 +50,18 @@ pub enum ChangeSetStatus { } impl ChangeSetStatus { + pub fn is_active_or_applying(&self) -> bool { + matches!( + self, + ChangeSetStatus::Open + | ChangeSetStatus::NeedsApproval + | ChangeSetStatus::NeedsAbandonApproval + | ChangeSetStatus::Approved + | ChangeSetStatus::Rejected + | ChangeSetStatus::ApplyStarted + ) + } + pub fn is_active(&self) -> bool { matches!( self, @@ -73,6 +87,7 @@ impl From for ChangeSetStatus { si_events::ChangeSetStatus::Open => Self::Open, si_events::ChangeSetStatus::Approved => Self::Approved, si_events::ChangeSetStatus::Rejected => Self::Rejected, + si_events::ChangeSetStatus::ApplyStarted => Self::ApplyStarted, } } } @@ -82,6 +97,7 @@ impl From for si_events::ChangeSetStatus { match value { ChangeSetStatus::Abandoned => Self::Abandoned, ChangeSetStatus::Applied => Self::Applied, + ChangeSetStatus::ApplyStarted => Self::ApplyStarted, ChangeSetStatus::Failed => Self::Failed, ChangeSetStatus::NeedsAbandonApproval => Self::NeedsAbandonApproval, ChangeSetStatus::NeedsApproval => Self::NeedsApproval, diff --git a/lib/dal/src/context.rs b/lib/dal/src/context.rs index 5d098b56d7..1f761d24b1 100644 --- a/lib/dal/src/context.rs +++ b/lib/dal/src/context.rs @@ -19,9 +19,13 @@ use futures::{ use rebaser_client::{ RebaserClient, RequestId, - api_types::enqueue_updates_response::{ - EnqueueUpdatesResponse, - RebaseStatus, + api_types::{ + enqueue_updates_request::ApplyToHeadRequestV4, + enqueue_updates_response::{ + EnqueueUpdatesResponse, + EnqueueUpdatesResponseVCurrent, + RebaseStatus, + }, }, }; use serde::{ @@ -617,6 +621,24 @@ impl DalContext { .await } + pub async fn run_async_apply_to_head( + &self, + request: ApplyToHeadRequestV4, + ) -> TransactionsResult { + self.rebaser() + .enqueue_apply_to_head( + request.workspace_id, + request.head_change_set_id, + request.head_change_set_address, + request.change_set_to_apply_id, + request.event_session_id, + request.previous_change_set_status, + request.mode, + ) + .await + .map_err(Into::into) + } + pub async fn run_async_rebase_from_change_set( &self, workspace_pk: WorkspacePk, @@ -625,7 +647,7 @@ impl DalContext { from_change_set_id: ChangeSetId, ) -> TransactionsResult { self.rebaser() - .enqueue_updates_from_change_set( + .enqueue_rebase_from_change_set( workspace_pk, change_set_id, updates_address, @@ -647,7 +669,7 @@ impl DalContext { BoxFuture<'static, Result>, )> { self.rebaser() - .enqueue_updates_from_change_set_with_reply( + .enqueue_rebase_from_change_set_with_reply( workspace_pk, change_set_id, updates_address, @@ -1014,10 +1036,10 @@ impl DalContext { Ok(()) } - /// Clones a new context from this one with a new [`Visibility`]. - pub fn clone_with_new_visibility(&self, visibility: Visibility) -> Self { + /// Clones a new context from this one with a new change set. + pub fn clone_with_change_set(&self, change_set_id: ChangeSetId) -> Self { let mut new = self.clone(); - new.update_visibility_deprecated(visibility); + new.update_visibility_deprecated(Visibility { change_set_id }); new } @@ -1250,9 +1272,7 @@ impl DalContext { } // Gets a reference to the DAL context's Rebaser client. - // - // **NOTE**: Internal API - fn rebaser(&self) -> &RebaserClient { + pub fn rebaser(&self) -> &RebaserClient { &self.services_context.rebaser } @@ -1705,6 +1725,8 @@ impl DalContextBuilder { #[remain::sorted] #[derive(Debug, Error, EnumDiscriminants)] pub enum TransactionsError { + #[error("apply to head {0} of change set {1} failed: {2}")] + ApplyFailed(ChangeSetId, ChangeSetId, String), #[error("audit logging error: {0}")] AuditLogging(#[from] Box), #[error("expected a {0:?} activity, but received a {1:?}")] @@ -1714,6 +1736,8 @@ pub enum TransactionsError { /// was specified. #[error("Bad Workspace & Change Set")] BadWorkspaceAndChangeSet, + #[error("Begin apply of {0} to head failed: {1}")] + BeginApplyFailed(ChangeSetId, String), #[error("change set error: {0}")] ChangeSet(#[from] Box), #[error("change set not set on DalContext")] @@ -1740,6 +1764,8 @@ pub enum TransactionsError { Rebaser(#[from] Box), #[error("rebaser reply deadline elapsed; waited={0:?}, request_id={1}")] RebaserReplyDeadlineElasped(Duration, RequestId), + #[error("rebaser unexpected response: {0:?}")] + RebaserUnexpectedResponse(EnqueueUpdatesResponse), #[error("serde json error: {0}")] SerdeJson(#[from] serde_json::Error), #[error("si db error: {0}")] @@ -2077,7 +2103,7 @@ async fn rebase_with_reply( let metric_label = format!("{workspace_pk}:{change_set_id}"); metric!(counter.dal.rebase_requested = 1, label = metric_label); let (request_id, reply_fut) = rebaser - .enqueue_updates_with_reply( + .enqueue_rebase_with_reply( workspace_pk, change_set_id, updates_address, @@ -2100,13 +2126,18 @@ async fn rebase_with_reply( metric!(counter.dal.rebase_requested = -1, label = metric_label); - match &reply.status { + let EnqueueUpdatesResponse::V2(EnqueueUpdatesResponseVCurrent::Rebase { status, .. }) = reply + else { + return Err(TransactionsError::RebaserUnexpectedResponse(reply.clone())); + }; + + match status { RebaseStatus::Success { .. } => Ok(()), // Return a specific error if the Rebaser reports that it failed to process the request RebaseStatus::Error { message } => Err(TransactionsError::RebaseFailed( updates_address, change_set_id, - message.clone(), + message, )), } } diff --git a/lib/dal/src/workspace_snapshot/migrator.rs b/lib/dal/src/workspace_snapshot/migrator.rs index de67be307d..7c59aea891 100644 --- a/lib/dal/src/workspace_snapshot/migrator.rs +++ b/lib/dal/src/workspace_snapshot/migrator.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; -use si_db::Visibility; use si_events::WorkspaceSnapshotAddress; use si_layer_cache::LayerDbError; use si_split_graph::SplitGraphError; @@ -119,8 +118,7 @@ impl SnapshotGraphMigrator { // on it, since its main purpose is to allow access to the services context. // We need to create a context for each migrated changeset here to run operations // that depend on the graph - let mut ctx_after_migration = - ctx.clone_with_new_visibility(Visibility::from(change_set.id)); + let mut ctx_after_migration = ctx.clone_with_change_set(change_set.id); // TODO make sure that when we clone with a changeset id we also set changeset // (or there's no clone anymore and we always change it via following method) ctx_after_migration.set_change_set(change_set.clone())?; diff --git a/lib/dal/tests/integration_test/action.rs b/lib/dal/tests/integration_test/action.rs index 2a3120532b..577b487c86 100644 --- a/lib/dal/tests/integration_test/action.rs +++ b/lib/dal/tests/integration_test/action.rs @@ -398,7 +398,7 @@ async fn refresh_actions_run_where_they_should(ctx: &mut DalContext) -> Result<( ChangeSetTestHelpers::fork_from_head_change_set(ctx).await?; Action::enqueue_refresh_in_correct_change_set_and_commit(ctx, small_even_lego.id()).await?; ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; - let seconds = 10; + let seconds = 120; let mut did_pass = false; for _ in 0..(seconds * 10) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; diff --git a/lib/dal/tests/integration_test/apply.rs b/lib/dal/tests/integration_test/apply.rs new file mode 100644 index 0000000000..1e7cf629f7 --- /dev/null +++ b/lib/dal/tests/integration_test/apply.rs @@ -0,0 +1,86 @@ +use dal::{ + ChangeSet, + Component, + DalContext, +}; +use dal_test::{ + Result, + helpers::{ + ChangeSetTestHelpers, + component, + }, + test, +}; +use pretty_assertions_sorted::assert_eq; + +// Creates three components on three different change sets. +// Then, by spawning three tasks, applies them to head concurrently, +// once all the tasks are complete, we assert that all three components +// exist on head. This test is very slow! +#[test] +async fn concurrent_applies_to_head(ctx: &mut DalContext) -> Result<()> { + // Create three components on three separate change sets + let component_1_id = component::create(ctx, "swifty", "component_1").await?; + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; + ChangeSet::wait_for_dvu(ctx, false).await?; + let change_set_1_id = ctx.change_set_id(); + + ChangeSetTestHelpers::fork_from_head_change_set(ctx).await?; + let component_2_id = component::create(ctx, "swifty", "component_2").await?; + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; + ChangeSet::wait_for_dvu(ctx, false).await?; + let change_set_2_id = ctx.change_set_id(); + + ChangeSetTestHelpers::fork_from_head_change_set(ctx).await?; + let component_3_id = component::create(ctx, "swifty", "component_3").await?; + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx).await?; + ChangeSet::wait_for_dvu(ctx, false).await?; + let change_set_3_id = ctx.change_set_id(); + + // Spawn three concurrent tasks to apply each change set to head + let handle_1 = { + let mut ctx_clone = ctx.clone_with_change_set(change_set_1_id); + tokio::spawn(async move { + ChangeSetTestHelpers::apply_change_set_to_base(&mut ctx_clone) + .await + .expect("failed to apply change set 1"); + }) + }; + + let handle_2 = { + let mut ctx_clone = ctx.clone_with_change_set(change_set_2_id); + tokio::spawn(async move { + ChangeSetTestHelpers::apply_change_set_to_base(&mut ctx_clone) + .await + .expect("failed to apply change set 2"); + }) + }; + + let handle_3 = { + let mut ctx_clone = ctx.clone_with_change_set(change_set_3_id); + tokio::spawn(async move { + ChangeSetTestHelpers::apply_change_set_to_base(&mut ctx_clone) + .await + .expect("failed to apply change set 3"); + }) + }; + + // Wait for all tasks to complete + handle_1.await.expect("task 1 panicked"); + handle_2.await.expect("task 2 panicked"); + handle_3.await.expect("task 3 panicked"); + + // Update context to HEAD and verify all three components exist + let ctx = ctx.clone_with_head().await?; + + let component_1 = Component::get_by_id(&ctx, component_1_id).await?; + assert_eq!("component_1", component_1.name(&ctx).await?); + + let component_2 = Component::get_by_id(&ctx, component_2_id).await?; + assert_eq!("component_2", component_2.name(&ctx).await?); + + let component_3 = Component::get_by_id(&ctx, component_3_id).await?; + assert_eq!("component_3", component_3.name(&ctx).await?); + + Ok(()) +} diff --git a/lib/dal/tests/integration_test/change_set.rs b/lib/dal/tests/integration_test/change_set.rs index 080109eb55..5e2606736f 100644 --- a/lib/dal/tests/integration_test/change_set.rs +++ b/lib/dal/tests/integration_test/change_set.rs @@ -430,7 +430,7 @@ async fn build_from_request_context_allows_change_set_from_workspace_with_access #[test] async fn change_set_approval_flow(ctx: &mut DalContext) { // create a new change set - let new_change_set = ChangeSetTestHelpers::fork_from_head_change_set(ctx) + let change_set = ChangeSetTestHelpers::fork_from_head_change_set(ctx) .await .expect("could not fork head"); let current_user = ChangeSet::extract_userid_from_context(ctx).await; @@ -444,7 +444,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { .await .expect("could not commit and update"); // request approval - let mut change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let mut change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); @@ -456,7 +456,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) .await .expect("could not commit and update"); - let mut change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let mut change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); @@ -475,7 +475,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) .await .expect("could not commit and update"); - let mut change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let mut change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); assert_eq!(change_set.status, ChangeSetStatus::Rejected); @@ -485,7 +485,8 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { assert_eq!(change_set.reviewed_by_user_id, current_user); // let's see if we can apply now, it should fail because the change set has not been approved - let apply_result = ChangeSetTestHelpers::apply_change_set_to_base_approvals(ctx).await; + let apply_result = + ChangeSetTestHelpers::apply_change_set_to_base_approvals_with_status_check(ctx).await; assert!(apply_result.is_err()); // now let's re-open it @@ -496,7 +497,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) .await .expect("could not commit and update"); - let mut change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let mut change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); assert_eq!(change_set.status, ChangeSetStatus::Open); @@ -514,7 +515,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) .await .expect("could not commit and update"); - let mut change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let mut change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); @@ -534,7 +535,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) .await .expect("could not commit and update"); - let change_set = ChangeSet::get_by_id(ctx, new_change_set.id) + let change_set = ChangeSet::get_by_id(ctx, change_set.id) .await .expect("could not find change set"); @@ -546,8 +547,7 @@ async fn change_set_approval_flow(ctx: &mut DalContext) { assert_eq!(change_set.reviewed_by_user_id, current_user); // now let's apply it! - - ChangeSetTestHelpers::apply_change_set_to_base_approvals(ctx) + ChangeSetTestHelpers::apply_change_set_to_base_approvals_with_status_check(ctx) .await .expect("could not apply to head"); diff --git a/lib/dal/tests/integration_test/management.rs b/lib/dal/tests/integration_test/management.rs index c1e59ffe1d..2e0315f494 100644 --- a/lib/dal/tests/integration_test/management.rs +++ b/lib/dal/tests/integration_test/management.rs @@ -2,6 +2,7 @@ use std::time::Duration; use dal::{ AttributeValue, + ChangeSet, Component, ComponentId, DalContext, @@ -268,7 +269,7 @@ async fn import_and_refresh(ctx: &mut DalContext) -> Result<()> { .and_then(|v| v.as_u64()) .expect("has a refresh_count"); assert_eq!(serde_json::json!(2), refresh_count); - + ChangeSet::wait_for_dvu(ctx, false).await?; // now apply the change set ChangeSetTestHelpers::apply_change_set_to_base(ctx).await?; // fork head, and run import again. Refresh should not run in this case. diff --git a/lib/dal/tests/integration_test/mod.rs b/lib/dal/tests/integration_test/mod.rs index f68a3df10a..db0e4f7192 100644 --- a/lib/dal/tests/integration_test/mod.rs +++ b/lib/dal/tests/integration_test/mod.rs @@ -1,4 +1,5 @@ mod action; +mod apply; mod asset; mod attribute_prototype; mod attribute_value; diff --git a/lib/dal/tests/integration_test/node_weight/schema_variant.rs b/lib/dal/tests/integration_test/node_weight/schema_variant.rs index b634058b0a..3cd8143f5c 100644 --- a/lib/dal/tests/integration_test/node_weight/schema_variant.rs +++ b/lib/dal/tests/integration_test/node_weight/schema_variant.rs @@ -98,7 +98,7 @@ async fn keeps_only_one_unlocked_variant_from_concurrent_requests(ctx: &mut DalC // NOTE: We *CANNOT* use `ChangeSetTestHelpers::apply_change_set_to_base` as it explicitly // locks all editing SchemaVariants, which defeats what we're attempting to test. - ChangeSetTestHelpers::apply_change_set_to_base_inner(ctx) + ChangeSetTestHelpers::apply_change_set_to_base_inner(ctx, true, false) .await .expect("Unable to apply first request to base change set"); @@ -126,7 +126,7 @@ async fn keeps_only_one_unlocked_variant_from_concurrent_requests(ctx: &mut DalC ctx.update_visibility_and_snapshot_to_visibility(second_request_change_set.id) .await .expect("Unable to update ctx to second request change set"); - ChangeSetTestHelpers::apply_change_set_to_base_inner(ctx) + ChangeSetTestHelpers::apply_change_set_to_base_inner(ctx, true, false) .await .expect("Unable to apply second request to base change set"); diff --git a/lib/dal/tests/integration_test/node_weight/view.rs b/lib/dal/tests/integration_test/node_weight/view.rs index ae6d598264..e38217816f 100644 --- a/lib/dal/tests/integration_test/node_weight/view.rs +++ b/lib/dal/tests/integration_test/node_weight/view.rs @@ -1,4 +1,7 @@ +use std::time::Duration; + use dal::{ + ChangeSet, Component, DalContext, diagram::{ @@ -18,7 +21,10 @@ use dal_test::{ generate_fake_name, }, helpers::create_component_for_default_schema_name, - prelude::OptionExt, + prelude::{ + ChangeSetTestHelpers, + OptionExt, + }, test, }; use petgraph::Direction::Outgoing; @@ -81,6 +87,22 @@ async fn correct_transforms_remove_view_all_geometries_removed(ctx: &mut DalCont View::list(ctx).await.expect("Unable to list Views").len() ); + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) + .await + .expect("Failed to commit and update snapshot to visibility"); + + let mut head_ctx = ctx + .clone_with_head() + .await + .expect("Failed to clone context"); + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU"); + tokio::time::sleep(Duration::from_secs(1)).await; + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU the 2nd time"); + expected::apply_change_set_to_base(ctx).await; assert_eq!( @@ -364,6 +386,23 @@ async fn correct_transforms_remove_view_with_component_in_another_view(ctx: &mut .len() ); + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) + .await + .expect("Failed to commit and update snapshot to visibility"); + + let mut head_ctx = ctx + .clone_with_head() + .await + .expect("Failed to clone context"); + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU"); + tokio::time::sleep(Duration::from_secs(1)).await; + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU the 2nd time"); + tokio::time::sleep(Duration::from_secs(1)).await; + expected::apply_change_set_to_base(ctx).await; assert_eq!( @@ -394,6 +433,15 @@ async fn correct_transforms_remove_view_components_moved_to_another_view( .await .expect("Unable to create Component in new View"); expected::apply_change_set_to_base(ctx).await; + + let mut head_ctx = ctx + .clone_with_head() + .await + .expect("Failed to clone context"); + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU"); + expected::fork_from_head_change_set(ctx).await; assert_eq!( @@ -426,7 +474,20 @@ async fn correct_transforms_remove_view_components_moved_to_another_view( .await? .len(), ); + + ChangeSetTestHelpers::commit_and_update_snapshot_to_visibility(ctx) + .await + .expect("Failed to commit and update snapshot to visibility"); + + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU"); + tokio::time::sleep(Duration::from_secs(5)).await; + ChangeSet::wait_for_dvu(&mut head_ctx, false) + .await + .expect("Failed to wait for DVU the 2nd time"); expected::apply_change_set_to_base(ctx).await; + tokio::time::sleep(Duration::from_secs(3)).await; // Applying the set of transforms should see that even though the View being removed // was the only one the Component was in, it is also being added to a different View diff --git a/lib/edda-server/src/change_set_processor_task.rs b/lib/edda-server/src/change_set_processor_task.rs index aeafc2c916..395413e959 100644 --- a/lib/edda-server/src/change_set_processor_task.rs +++ b/lib/edda-server/src/change_set_processor_task.rs @@ -422,7 +422,7 @@ mod handlers { // We should skip any change sets that are not active // Active in this case is open, needs approval status, approved or rejected - if !to_process_change_set.status.is_active() { + if !to_process_change_set.status.is_active_or_applying() { debug!("Attempted to process a non-active change set. Skipping"); return Ok(()); } diff --git a/lib/luminork-server/src/service/v1/change_sets/force_apply.rs b/lib/luminork-server/src/service/v1/change_sets/force_apply.rs index 4f5debe44e..782e91610c 100644 --- a/lib/luminork-server/src/service/v1/change_sets/force_apply.rs +++ b/lib/luminork-server/src/service/v1/change_sets/force_apply.rs @@ -48,7 +48,7 @@ pub async fn force_apply( } let old_status = ctx.change_set()?.status; - ChangeSet::prepare_for_force_apply(ctx).await?; + ChangeSet::force_change_set_approval(ctx).await?; ctx.write_audit_log( AuditLogKind::ApproveChangeSetApply { from_status: old_status.into(), @@ -59,7 +59,7 @@ pub async fn force_apply( ctx.commit().await?; - ChangeSet::apply_to_base_change_set(ctx).await?; + ChangeSet::begin_apply(ctx).await?; tracker.track( ctx, diff --git a/lib/luminork-server/src/service/v1/change_sets/request_approval.rs b/lib/luminork-server/src/service/v1/change_sets/request_approval.rs index 85910f0189..9b74a92835 100644 --- a/lib/luminork-server/src/service/v1/change_sets/request_approval.rs +++ b/lib/luminork-server/src/service/v1/change_sets/request_approval.rs @@ -87,7 +87,7 @@ pub async fn request_approval( .publish_on_commit(ctx) .await?; } else { - ChangeSet::prepare_for_force_apply(ctx).await?; + ChangeSet::force_change_set_approval(ctx).await?; ctx.write_audit_log( AuditLogKind::ApproveChangeSetApply { from_status: old_status.into(), @@ -97,9 +97,9 @@ pub async fn request_approval( .await?; // We need to run a commit before apply so changes get saved - ctx.commit().await?; + ctx.commit_no_rebase().await?; - ChangeSet::apply_to_base_change_set(ctx).await?; + ChangeSet::begin_apply(ctx).await?; tracker.track( ctx, diff --git a/lib/rebaser-client/src/lib.rs b/lib/rebaser-client/src/lib.rs index 9185f16bab..e12050171e 100644 --- a/lib/rebaser-client/src/lib.rs +++ b/lib/rebaser-client/src/lib.rs @@ -23,8 +23,12 @@ use rebaser_core::{ SerializeContainer, SerializeError, enqueue_updates_request::{ + ApplyToHeadRequestV4, + BeginApplyMode, + BeginApplyToHeadRequestV4, EnqueueUpdatesRequest, EnqueueUpdatesRequestVCurrent, + RebaseRequestV4, }, enqueue_updates_response::EnqueueUpdatesResponse, }, @@ -46,9 +50,11 @@ use si_data_nats::{ }; use si_events::{ ChangeSetId, + ChangeSetStatus, EventSessionId, RebaseBatchAddressKind, WorkspacePk, + WorkspaceSnapshotAddress, }; use telemetry::prelude::*; use telemetry_nats::propagation; @@ -106,7 +112,7 @@ impl Client { /// Asynchronously enqueues graph updates for processing by a Rebaser & return a [`RequestId`]. #[instrument( - name = "rebaser_client.enqueue_updates", + name = "rebaser_client.enqueue_rebase", level = "info", skip_all, fields( @@ -114,28 +120,32 @@ impl Client { si.workspace.id = %workspace_id, ), )] - pub async fn enqueue_updates( + pub async fn enqueue_rebase( &self, workspace_id: WorkspacePk, change_set_id: ChangeSetId, updates_address: RebaseBatchAddressKind, event_session_id: EventSessionId, ) -> Result { - self.call_async( - workspace_id, - change_set_id, - updates_address, - None, - None, - event_session_id, - ) - .await + let id = RequestId::new(); + + let request = EnqueueUpdatesRequestVCurrent::Rebase { + id, + request: RebaseRequestV4 { + workspace_id, + change_set_id, + updates_address, + from_change_set_id: None, + event_session_id: Some(event_session_id), + }, + }; + self.call_async(request, None, event_session_id).await } /// Asynchronously enqueues graph updates that originate from a Change Set & return a /// [`RequestId`]. #[instrument( - name = "rebaser_client.enqueue_updates_from_change_set", + name = "rebaser_client.enqueue_rebase_from_change_set", level = "info", skip_all, fields( @@ -143,7 +153,7 @@ impl Client { si.workspace.id = %workspace_id, ), )] - pub async fn enqueue_updates_from_change_set( + pub async fn enqueue_rebase_from_change_set( &self, workspace_id: WorkspacePk, change_set_id: ChangeSetId, @@ -151,21 +161,26 @@ impl Client { from_change_set_id: ChangeSetId, event_session_id: EventSessionId, ) -> Result { - self.call_async( - workspace_id, - change_set_id, - updates_address, - Some(from_change_set_id), - None, - event_session_id, - ) - .await + let id = RequestId::new(); + + let request = EnqueueUpdatesRequestVCurrent::Rebase { + id, + request: RebaseRequestV4 { + workspace_id, + change_set_id, + updates_address, + from_change_set_id: Some(from_change_set_id), + event_session_id: Some(event_session_id), + }, + }; + + self.call_async(request, None, event_session_id).await } /// Enqueues graph updates for processing by a Rebaser and return a [`Future`] that will await /// the Rebaser's response with status. #[instrument( - name = "rebaser_client.enqueue_updates_with_reply", + name = "rebaser_client.enqueue_rebase_with_reply", level = "info", skip_all, fields( @@ -173,7 +188,7 @@ impl Client { si.workspace.id = %workspace_id, ), )] - pub async fn enqueue_updates_with_reply( + pub async fn enqueue_rebase_with_reply( &self, workspace_id: WorkspacePk, change_set_id: ChangeSetId, @@ -183,20 +198,25 @@ impl Client { RequestId, BoxFuture<'static, Result>, )> { - self.call_with_reply( - workspace_id, - change_set_id, - updates_address, - None, - event_session_id, - ) - .await + let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::Rebase { + id, + request: RebaseRequestV4 { + workspace_id, + change_set_id, + updates_address, + from_change_set_id: None, + event_session_id: Some(event_session_id), + }, + }; + + self.call_with_reply(request, event_session_id).await } /// Enqueues graph updates that originate from a Change Set and return a [`Future`] that will /// await the Rebaser's response with status. #[instrument( - name = "rebaser_client.enqueue_updates_from_change_set_with_reply", + name = "rebaser_client.enqueue_rebase_from_change_set_with_reply", level = "info", skip_all, fields( @@ -204,7 +224,7 @@ impl Client { si.workspace.id = %workspace_id, ), )] - pub async fn enqueue_updates_from_change_set_with_reply( + pub async fn enqueue_rebase_from_change_set_with_reply( &self, workspace_id: WorkspacePk, change_set_id: ChangeSetId, @@ -215,26 +235,198 @@ impl Client { RequestId, BoxFuture<'static, Result>, )> { - self.call_with_reply( - workspace_id, - change_set_id, - updates_address, - Some(from_change_set_id), - event_session_id, - ) - .await + let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::Rebase { + id, + request: RebaseRequestV4 { + workspace_id, + change_set_id, + updates_address, + from_change_set_id: Some(from_change_set_id), + event_session_id: Some(event_session_id), + }, + }; + + self.call_with_reply(request, event_session_id).await } - async fn call_async( + /// Enqueues a request to begin applying a change set to the HEAD of a workspace. + #[instrument( + name = "rebaser_client.enqueue_begin_apply", + level = "info", + skip_all, + fields( + si.change_set.id = %change_set_id, + si.workspace.id = %workspace_id, + si.head_change_set.id = %head_change_set_id, + ), + )] + #[allow(clippy::too_many_arguments)] + pub async fn enqueue_begin_apply( &self, workspace_id: WorkspacePk, change_set_id: ChangeSetId, - updates_address: RebaseBatchAddressKind, - from_change_set_id: Option, - maybe_reply_inbox: Option<&Subject>, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + previous_change_set_status: Option, + event_session_id: EventSessionId, + mode: BeginApplyMode, + ) -> Result { + let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::BeginApplyToHead { + id, + request: BeginApplyToHeadRequestV4 { + workspace_id, + change_set_id, + head_change_set_id, + head_change_set_address, + event_session_id, + previous_change_set_status, + mode, + }, + }; + + self.call_async(request, None, event_session_id).await + } + + /// Enqueues a request to begin applying a change set to the HEAD of a + /// workspace, returning a reply future. + #[instrument( + name = "rebaser_client.enqueue_begin_apply_with_reply", + level = "info", + skip_all, + fields( + si.change_set.id = %change_set_id, + si.workspace.id = %workspace_id, + si.head_change_set.id = %head_change_set_id, + ), + )] + #[allow(clippy::too_many_arguments)] + pub async fn enqueue_begin_apply_with_retry( + &self, + workspace_id: WorkspacePk, + change_set_id: ChangeSetId, + previous_change_set_status: Option, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + event_session_id: EventSessionId, + mode: BeginApplyMode, + ) -> Result<( + RequestId, + BoxFuture<'static, Result>, + )> { + let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::BeginApplyToHead { + id, + request: BeginApplyToHeadRequestV4 { + workspace_id, + change_set_id, + previous_change_set_status, + head_change_set_address, + head_change_set_id, + event_session_id, + mode, + }, + }; + + self.call_with_reply(request, event_session_id).await + } + + /// Enqueues a request to apply a change set to the head + /// workspace. + #[instrument( + name = "rebaser_client.apply_to_head", + level = "info", + skip_all, + fields( + si.change_set.id = %head_change_set_id, + si.workspace.id = %workspace_id, + si.change_set_to_apply.id= %change_set_to_apply_id, + ), + )] + #[allow(clippy::too_many_arguments)] + pub async fn enqueue_apply_to_head( + &self, + workspace_id: WorkspacePk, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + change_set_to_apply_id: ChangeSetId, event_session_id: EventSessionId, + previous_change_set_status: ChangeSetStatus, + mode: Option, ) -> Result { let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::ApplyToHead { + id, + request: ApplyToHeadRequestV4 { + workspace_id, + head_change_set_id, + head_change_set_address, + change_set_to_apply_id, + event_session_id, + previous_change_set_status, + mode, + }, + }; + + self.call_async(request, None, event_session_id).await + } + + /// Enqueues a request to apply a change set to the head + /// workspace, with reply. + #[instrument( + name = "rebaser_client.apply_to_head_with_reply", + level = "info", + skip_all, + fields( + si.change_set.id = %head_change_set_id, + si.workspace.id = %workspace_id, + si.change_set_to_apply.id= %change_set_to_apply_id, + ), + )] + #[allow(clippy::too_many_arguments)] + pub async fn enqueue_apply_to_head_with_reply( + &self, + workspace_id: WorkspacePk, + head_change_set_id: ChangeSetId, + head_change_set_address: WorkspaceSnapshotAddress, + change_set_to_apply_id: ChangeSetId, + event_session_id: EventSessionId, + previous_change_set_status: ChangeSetStatus, + mode: Option, + ) -> Result<( + RequestId, + BoxFuture<'static, Result>, + )> { + let id = RequestId::new(); + let request = EnqueueUpdatesRequestVCurrent::ApplyToHead { + id, + request: ApplyToHeadRequestV4 { + workspace_id, + head_change_set_id, + head_change_set_address, + change_set_to_apply_id, + event_session_id, + previous_change_set_status, + mode, + }, + }; + + self.call_with_reply(request, event_session_id).await + } + + async fn call_async( + &self, + mut request: EnqueueUpdatesRequestVCurrent, + maybe_reply_inbox: Option<&Subject>, + event_session_id: EventSessionId, + ) -> Result { + let id = request.request_id(); + + let workspace_id = request.workspace_id(); + let change_set_id = request.change_set_id(); + + request.set_event_session_id(event_session_id); let pending_events_stream = PendingEventsStream::get_or_create(self.context.to_owned()).await?; @@ -242,14 +434,7 @@ impl Client { .publish_audit_log_final_message(workspace_id, change_set_id, event_session_id) .await?; - let request = EnqueueUpdatesRequest::new(EnqueueUpdatesRequestVCurrent { - id, - workspace_id, - change_set_id, - updates_address, - from_change_set_id, - event_session_id: Some(event_session_id), - }); + let request = EnqueueUpdatesRequest::new(request); // Cut down on the amount of `String` allocations dealing with ids let mut wid_buf = [0; WorkspacePk::ID_LEN]; @@ -293,10 +478,7 @@ impl Client { async fn call_with_reply( &self, - workspace_id: WorkspacePk, - change_set_id: ChangeSetId, - updates_address: RebaseBatchAddressKind, - from_change_set_id: Option, + request: EnqueueUpdatesRequestVCurrent, event_session_id: EventSessionId, ) -> Result<( RequestId, @@ -319,14 +501,7 @@ impl Client { .map_err(Error::Subscribe)?; let id = self - .call_async( - workspace_id, - change_set_id, - updates_address, - from_change_set_id, - Some(&reply_inbox), - event_session_id, - ) + .call_async(request, Some(&reply_inbox), event_session_id) .await?; let fut = Box::pin(async move { diff --git a/lib/rebaser-core/src/api_types/enqueue_updates_request.rs b/lib/rebaser-core/src/api_types/enqueue_updates_request.rs index 68e199e57f..8f7624a440 100644 --- a/lib/rebaser-core/src/api_types/enqueue_updates_request.rs +++ b/lib/rebaser-core/src/api_types/enqueue_updates_request.rs @@ -11,11 +11,23 @@ use si_events::RebaseBatchAddressKind; mod v1; mod v2; mod v3; +mod v4; + +pub use acceptable::RequestId; pub use self::{ v1::EnqueueUpdatesRequestV1, v2::EnqueueUpdatesRequestV2, v3::EnqueueUpdatesRequestV3, + v4::{ + ApplyToHeadRequestV4, + BeginApplyMode, + BeginApplyToHeadRequestV4, + ChangeSetStatusUpdateRequestV4, + CreateChangeSetRequestV4, + EnqueueUpdatesRequestV4, + RebaseRequestV4, + }, }; #[remain::sorted] @@ -24,8 +36,9 @@ pub use self::{ pub enum EnqueueUpdatesRequestAllVersions { V1(EnqueueUpdatesRequestV1), V2(EnqueueUpdatesRequestV2), - #[acceptable(current)] V3(EnqueueUpdatesRequestV3), + #[acceptable(current)] + V4(EnqueueUpdatesRequestV4), } impl IntoContainer for EnqueueUpdatesRequestAllVersions { @@ -33,23 +46,37 @@ impl IntoContainer for EnqueueUpdatesRequestAllVersions { fn into_container(self) -> Result { Ok(match self { - Self::V1(inner) => Self::Container::new(EnqueueUpdatesRequestVCurrent { + Self::V1(inner) => Self::Container::new(EnqueueUpdatesRequestVCurrent::Rebase { + id: inner.id, + request: RebaseRequestV4 { + workspace_id: inner.workspace_id, + change_set_id: inner.change_set_id, + updates_address: RebaseBatchAddressKind::Legacy(inner.updates_address), + from_change_set_id: inner.from_change_set_id, + event_session_id: None, + }, + }), + Self::V2(inner) => Self::Container::new(EnqueueUpdatesRequestVCurrent::Rebase { id: inner.id, - workspace_id: inner.workspace_id, - change_set_id: inner.change_set_id, - updates_address: RebaseBatchAddressKind::Legacy(inner.updates_address), - from_change_set_id: inner.from_change_set_id, - event_session_id: None, + request: RebaseRequestV4 { + workspace_id: inner.workspace_id, + change_set_id: inner.change_set_id, + updates_address: RebaseBatchAddressKind::Legacy(inner.updates_address), + from_change_set_id: inner.from_change_set_id, + event_session_id: inner.event_session_id, + }, }), - Self::V2(inner) => Self::Container::new(EnqueueUpdatesRequestVCurrent { + Self::V3(inner) => Self::Container::new(EnqueueUpdatesRequestVCurrent::Rebase { id: inner.id, - workspace_id: inner.workspace_id, - change_set_id: inner.change_set_id, - updates_address: RebaseBatchAddressKind::Legacy(inner.updates_address), - from_change_set_id: inner.from_change_set_id, - event_session_id: inner.event_session_id, + request: RebaseRequestV4 { + workspace_id: inner.workspace_id, + change_set_id: inner.change_set_id, + updates_address: inner.updates_address, + from_change_set_id: inner.from_change_set_id, + event_session_id: inner.event_session_id, + }, }), - Self::V3(inner) => Self::Container::new(inner), + Self::V4(inner) => Self::Container::new(inner), }) } } diff --git a/lib/rebaser-core/src/api_types/enqueue_updates_request/v4.rs b/lib/rebaser-core/src/api_types/enqueue_updates_request/v4.rs new file mode 100644 index 0000000000..a96f134401 --- /dev/null +++ b/lib/rebaser-core/src/api_types/enqueue_updates_request/v4.rs @@ -0,0 +1,160 @@ +use acceptable::{ + RequestId, + Versioned, +}; +use serde::{ + Deserialize, + Serialize, +}; +use si_events::{ + ChangeSetId, + ChangeSetStatus, + EventSessionId, + RebaseBatchAddressKind, + WorkspacePk, + WorkspaceSnapshotAddress, +}; + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq, Versioned)] +#[serde(rename_all = "camelCase")] +#[acceptable(version = 4)] +pub enum EnqueueUpdatesRequestV4 { + Rebase { + id: RequestId, + request: RebaseRequestV4, + }, + BeginApplyToHead { + id: RequestId, + request: BeginApplyToHeadRequestV4, + }, + ApplyToHead { + id: RequestId, + request: ApplyToHeadRequestV4, + }, + CreateChangeSet { + id: RequestId, + request: CreateChangeSetRequestV4, + }, + ChangeSetStatusUpdate { + id: RequestId, + request: ChangeSetStatusUpdateRequestV4, + }, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct RebaseRequestV4 { + pub workspace_id: WorkspacePk, + pub change_set_id: ChangeSetId, + pub updates_address: RebaseBatchAddressKind, + pub from_change_set_id: Option, + pub event_session_id: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub enum BeginApplyMode { + LockSchemasAndFuncs, + NoLock, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct BeginApplyToHeadRequestV4 { + pub workspace_id: WorkspacePk, + pub change_set_id: ChangeSetId, + pub head_change_set_id: ChangeSetId, + pub head_change_set_address: WorkspaceSnapshotAddress, + pub event_session_id: EventSessionId, + pub previous_change_set_status: Option, + pub mode: BeginApplyMode, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct ApplyToHeadRequestV4 { + pub workspace_id: WorkspacePk, + pub head_change_set_id: ChangeSetId, + pub head_change_set_address: WorkspaceSnapshotAddress, + pub change_set_to_apply_id: ChangeSetId, + pub event_session_id: EventSessionId, + pub previous_change_set_status: ChangeSetStatus, + pub mode: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct CreateChangeSetRequestV4 { + pub workspace_id: WorkspacePk, + pub head_change_set_id: ChangeSetId, + pub name: String, + pub event_session_id: EventSessionId, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct ChangeSetStatusUpdateRequestV4 { + pub workspace_id: WorkspacePk, + pub change_set_id: ChangeSetId, + pub status: ChangeSetStatus, + pub event_session_id: EventSessionId, +} + +impl EnqueueUpdatesRequestV4 { + pub fn request_id(&self) -> RequestId { + match self { + EnqueueUpdatesRequestV4::Rebase { id, .. } => *id, + EnqueueUpdatesRequestV4::BeginApplyToHead { id, .. } => *id, + EnqueueUpdatesRequestV4::ApplyToHead { id, .. } => *id, + EnqueueUpdatesRequestV4::CreateChangeSet { id, .. } => *id, + EnqueueUpdatesRequestV4::ChangeSetStatusUpdate { id, .. } => *id, + } + } + + /// This is the workspace id that will be used for the NATs subject + pub fn workspace_id(&self) -> WorkspacePk { + match self { + EnqueueUpdatesRequestV4::Rebase { request, .. } => request.workspace_id, + EnqueueUpdatesRequestV4::BeginApplyToHead { request, .. } => request.workspace_id, + EnqueueUpdatesRequestV4::ApplyToHead { request, .. } => request.workspace_id, + EnqueueUpdatesRequestV4::CreateChangeSet { request, .. } => request.workspace_id, + EnqueueUpdatesRequestV4::ChangeSetStatusUpdate { request, .. } => request.workspace_id, + } + } + + /// The change set id that will be used for the NATs subject + pub fn change_set_id(&self) -> ChangeSetId { + match self { + EnqueueUpdatesRequestV4::Rebase { request, .. } => request.change_set_id, + EnqueueUpdatesRequestV4::BeginApplyToHead { request, .. } => request.change_set_id, + EnqueueUpdatesRequestV4::ApplyToHead { request, .. } => request.head_change_set_id, + EnqueueUpdatesRequestV4::CreateChangeSet { request, .. } => request.head_change_set_id, + EnqueueUpdatesRequestV4::ChangeSetStatusUpdate { request, .. } => request.change_set_id, + } + } + + /// Set event session id if not set + pub fn set_event_session_id(&mut self, event_session_id: EventSessionId) { + if let EnqueueUpdatesRequestV4::Rebase { request, .. } = self { + if request.event_session_id.is_none() { + request.event_session_id = Some(event_session_id); + } + } + } + + pub fn event_session_id(&self) -> Option { + match self { + EnqueueUpdatesRequestV4::Rebase { request, .. } => request.event_session_id, + EnqueueUpdatesRequestV4::BeginApplyToHead { request, .. } => { + Some(request.event_session_id) + } + EnqueueUpdatesRequestV4::ApplyToHead { request, .. } => Some(request.event_session_id), + EnqueueUpdatesRequestV4::CreateChangeSet { request, .. } => { + Some(request.event_session_id) + } + EnqueueUpdatesRequestV4::ChangeSetStatusUpdate { request, .. } => { + Some(request.event_session_id) + } + } + } +} diff --git a/lib/rebaser-core/src/api_types/enqueue_updates_response.rs b/lib/rebaser-core/src/api_types/enqueue_updates_response.rs index bca1c73553..563cc31437 100644 --- a/lib/rebaser-core/src/api_types/enqueue_updates_response.rs +++ b/lib/rebaser-core/src/api_types/enqueue_updates_response.rs @@ -8,18 +8,29 @@ use acceptable::{ use serde::Deserialize; mod v1; +mod v2; -pub use self::v1::{ - EnqueueUpdatesResponseV1, - RebaseStatus, +pub use self::{ + v1::{ + EnqueueUpdatesResponseV1, + RebaseStatus, + }, + v2::{ + ApplyToHeadStatus, + BeginApplyStatus, + CreateChangeSetStatus, + EnqueueUpdatesResponseV2, + UpdateChangeSetStatusStatus, + }, }; #[remain::sorted] #[derive(AllVersions, CurrentContainer, Clone, Debug, Deserialize, Eq, PartialEq)] #[serde(rename_all = "camelCase")] pub enum EnqueueUpdatesResponseAllVersions { - #[acceptable(current)] V1(EnqueueUpdatesResponseV1), + #[acceptable(current)] + V2(EnqueueUpdatesResponseV2), } impl IntoContainer for EnqueueUpdatesResponseAllVersions { @@ -27,7 +38,13 @@ impl IntoContainer for EnqueueUpdatesResponseAllVersions { fn into_container(self) -> Result { match self { - Self::V1(inner) => Ok(Self::Container::new(inner)), + Self::V1(inner) => Ok(Self::Container::new(EnqueueUpdatesResponseV2::Rebase { + id: inner.id, + workspace_id: inner.workspace_id, + change_set_id: inner.change_set_id, + status: inner.status, + })), + Self::V2(inner) => Ok(Self::Container::new(inner)), } } } diff --git a/lib/rebaser-core/src/api_types/enqueue_updates_response/v2.rs b/lib/rebaser-core/src/api_types/enqueue_updates_response/v2.rs new file mode 100644 index 0000000000..dd03ecd129 --- /dev/null +++ b/lib/rebaser-core/src/api_types/enqueue_updates_response/v2.rs @@ -0,0 +1,90 @@ +use acceptable::{ + RequestId, + Versioned, +}; +use serde::{ + Deserialize, + Serialize, +}; +use si_events::{ + ChangeSetId, + ChangeSetStatus, + WorkspacePk, +}; + +use super::v1::RebaseStatus; + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq, Versioned)] +#[serde(rename_all = "camelCase")] +#[acceptable(version = 2)] +pub enum EnqueueUpdatesResponseV2 { + Rebase { + id: RequestId, + workspace_id: WorkspacePk, + change_set_id: ChangeSetId, + status: RebaseStatus, + }, + BeginApplyToHead { + id: RequestId, + workspace_id: WorkspacePk, + change_set_id: ChangeSetId, + status: BeginApplyStatus, + }, + ApplyToHead { + id: RequestId, + workspace_id: WorkspacePk, + change_set_id: ChangeSetId, + status: ApplyToHeadStatus, + }, + CreateChangeSet { + id: RequestId, + workspace_id: WorkspacePk, + status: CreateChangeSetStatus, + }, + UpdateChangeSetStatus { + id: RequestId, + workspace_id: WorkspacePk, + change_set_id: ChangeSetId, + status: UpdateChangeSetStatusStatus, + }, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub enum BeginApplyStatus { + Success { + previous_change_set_status: ChangeSetStatus, + }, + Error { + message: String, + }, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub enum CreateChangeSetStatus { + Success { new_change_set_id: ChangeSetId }, + Error { message: String }, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub enum UpdateChangeSetStatusStatus { + Retry { + previous_change_set_status: ChangeSetStatus, + }, + Success { + previous_change_set_status: ChangeSetStatus, + }, + Error { + message: String, + }, +} + +#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub enum ApplyToHeadStatus { + Success, + Retrying, + Error { message: String }, +} diff --git a/lib/rebaser-server/src/apply.rs b/lib/rebaser-server/src/apply.rs new file mode 100644 index 0000000000..465a360fa9 --- /dev/null +++ b/lib/rebaser-server/src/apply.rs @@ -0,0 +1,265 @@ +use std::time::Duration; + +use dal::{ + ChangeSet, + ChangeSetId, + ChangeSetStatus, + DalContext, + TransactionsError, + WorkspaceSnapshotAddress, + WorkspaceSnapshotError, + WsEvent, + WsEventError, + workspace_snapshot::{ + DependentValueRoot, + dependent_value_root::DependentValueRootError, + selector::WorkspaceSnapshotSelectorDiscriminants, + }, +}; +use rebaser_core::api_types::enqueue_updates_request::{ + ApplyToHeadRequestV4, + BeginApplyMode, + BeginApplyToHeadRequestV4, + RebaseRequestV4, +}; +use si_events::RebaseBatchAddressKind; +use telemetry::prelude::*; +use thiserror::Error; +use tokio_util::task::TaskTracker; + +#[derive(Debug, Error)] +pub(crate) enum ApplyError { + #[error("change set error: {0}")] + ChangeSet(#[from] dal::ChangeSetError), + #[error("change set apply error: {0}")] + ChangeSetApply(#[from] dal::ChangeSetApplyError), + #[error("change set {0} not marked for apply: {1:?}")] + ChangeSetNotMarkedForApply(ChangeSetId, ChangeSetStatus), + #[error("change set {0} not open for apply: {1:?}")] + ChangeSetNotOpenForApply(ChangeSetId, ChangeSetStatus), + #[error("dependent value root error: {0}")] + DependentValueRoot(#[from] DependentValueRootError), + #[error("dependent value execution still in progress, try applying again when finished")] + DependentValueRootExists(ChangeSetStatus), + #[error("rebaser client error: {0}")] + RebaserClient(#[from] rebaser_client::ClientError), + #[error("transaction error: {0}")] + Transactions(#[from] TransactionsError), + #[error("workspace snapshot error: {0}")] + WorkspaceSnapshot(#[from] WorkspaceSnapshotError), + #[error("ws event error: {0}")] + WsEvent(#[from] WsEventError), +} + +pub(crate) type Result = std::result::Result; + +#[instrument(name = "rebaser.begin_apply", level = "info", skip_all)] +pub(crate) async fn begin_apply( + ctx: &mut DalContext, + request: &BeginApplyToHeadRequestV4, +) -> Result { + let mut change_set = ChangeSet::get_by_id(ctx, request.change_set_id).await?; + + let previous_change_set_status = if change_set.status == ChangeSetStatus::ApplyStarted + && request.previous_change_set_status.is_some() + { + request + .previous_change_set_status + .expect("we know it is some") + } else { + change_set.status.into() + }; + + if !change_set.status.is_active_or_applying() { + return Err(ApplyError::ChangeSetNotOpenForApply( + change_set.id, + change_set.status, + )); + } + + ctx.update_snapshot_to_visibility().await?; + + // Fail safe here in case we have raced the DVU + if DependentValueRoot::roots_exist(ctx).await? { + return Err(ApplyError::DependentValueRootExists( + previous_change_set_status.into(), + )); + } + + if request.mode == BeginApplyMode::LockSchemasAndFuncs { + ChangeSet::lock_schemas_and_funcs_for_apply(ctx).await?; + } + + // A rebase is not required here, since the rebaser by definition has the + // latest version of the snapshot. Just lock the schemas/funcs and write it + // out. + let new_address = ctx.workspace_snapshot()?.write(ctx).await?; + + change_set.update_pointer(ctx, new_address).await?; + change_set + .update_status(ctx, ChangeSetStatus::ApplyStarted) + .await?; + ctx.commit_no_rebase().await?; + + ChangeSet::apply_to_base_change_set( + ctx, + previous_change_set_status.into(), + request.head_change_set_address, + Some(request.mode), + ) + .await?; + + Ok(previous_change_set_status.into()) +} + +#[instrument(name = "rebaser.get_apply_rebase_request", level = "info", skip_all)] +pub(crate) async fn get_apply_rebase_request( + ctx: &DalContext, + request: ApplyToHeadRequestV4, +) -> Result> { + let ApplyToHeadRequestV4 { + workspace_id, + head_change_set_id, + change_set_to_apply_id, + event_session_id, + .. + } = request; + + let change_set = ChangeSet::get_by_id(ctx, change_set_to_apply_id).await?; + // Being marked as "ApplyStarted" means: schemas and funcs are locked, and + // change set has no DVU roots. + if change_set.status != ChangeSetStatus::ApplyStarted { + return Err(ApplyError::ChangeSetNotMarkedForApply( + change_set_to_apply_id, + change_set.status, + )); + } + + let mut ctx = ctx.clone(); + ctx.update_visibility_and_snapshot_to_visibility(change_set_to_apply_id) + .await?; + + let snapshot_kind: WorkspaceSnapshotSelectorDiscriminants = ctx.workspace_snapshot()?.into(); + + let maybe_rebase_batch_address = match snapshot_kind { + WorkspaceSnapshotSelectorDiscriminants::LegacySnapshot => { + if let Some(rebase_batch) = change_set + .detect_updates_that_will_be_applied_legacy(&ctx) + .await? + { + Some(RebaseBatchAddressKind::Legacy( + ctx.write_legacy_rebase_batch(rebase_batch).await?, + )) + } else { + None + } + } + WorkspaceSnapshotSelectorDiscriminants::SplitSnapshot => { + if let Some(rebase_batch) = change_set + .detect_updates_that_will_be_applied_split(&ctx) + .await? + { + Some(RebaseBatchAddressKind::Split( + ctx.write_split_snapshot_rebase_batch(rebase_batch).await?, + )) + } else { + None + } + } + }; + + Ok( + maybe_rebase_batch_address.map(|updates_address| RebaseRequestV4 { + workspace_id, + change_set_id: head_change_set_id, + updates_address, + from_change_set_id: Some(change_set_to_apply_id), + event_session_id: Some(event_session_id), + }), + ) +} + +#[instrument(name = "rebaser.mark_change_set_applied", level = "info", skip_all)] +pub(crate) async fn mark_change_set_applied( + ctx: &DalContext, + change_set_id: ChangeSetId, + head_change_set_id: ChangeSetId, +) -> Result<()> { + let mut change_set = ChangeSet::get_by_id(ctx, change_set_id).await?; + if change_set.status == ChangeSetStatus::Applied { + return Ok(()); + } + + change_set + .update_status(ctx, ChangeSetStatus::Applied) + .await?; + let user = ChangeSet::extract_userid_from_context(ctx).await; + + WsEvent::change_set_applied(ctx, change_set_id, head_change_set_id, user) + .await? + .publish_on_commit(ctx) + .await?; + + ctx.commit_no_rebase().await?; + + Ok(()) +} + +#[instrument(name = "rebaser.enqueue_apply_retry", level = "info", skip(ctx))] +pub(crate) async fn enqueue_apply_retry( + ctx: &DalContext, + request: ApplyToHeadRequestV4, + head_change_set_address: WorkspaceSnapshotAddress, +) -> Result<()> { + ctx.rebaser() + .enqueue_begin_apply( + request.workspace_id, + request.change_set_to_apply_id, + request.head_change_set_id, + head_change_set_address, + Some(request.previous_change_set_status), + request.event_session_id, + request.mode.unwrap_or(BeginApplyMode::LockSchemasAndFuncs), + ) + .await?; + + Ok(()) +} + +const DEFAULT_DELAY: Duration = Duration::from_millis(200); + +/// Enqueues an apply retry with a delay. When we retry because the DVU roots +/// exist, we don't want to retry immediately, since this could lead to us +/// spamming the change set and HEAD request streams with retry messages as fast +/// as we can process them until the DVU is finished. Instead wait 200ms and +/// then enqueue, to throttle the rate of retries. +#[instrument(name = "rebaser.enqueue_apply_retry", level = "info", skip(ctx))] +pub(crate) async fn enqueue_apply_retry_with_delay( + ctx: &DalContext, + request: ApplyToHeadRequestV4, + head_change_set_address: WorkspaceSnapshotAddress, + server_tracker: TaskTracker, +) { + let ctx = ctx.clone(); + server_tracker.spawn(async move { + tokio::time::sleep(DEFAULT_DELAY).await; + if let Err(err) = ctx + .rebaser() + .enqueue_begin_apply( + request.workspace_id, + request.change_set_to_apply_id, + request.head_change_set_id, + head_change_set_address, + Some(request.previous_change_set_status), + request.event_session_id, + request.mode.unwrap_or(BeginApplyMode::LockSchemasAndFuncs), + ) + .await + { + error!( + "Failed to enqueue begin apply {:?} with delay: {}", + request, err + ); + } + }); +} diff --git a/lib/rebaser-server/src/change_set_processor_task.rs b/lib/rebaser-server/src/change_set_processor_task.rs index d7f74fa3a9..8239225e30 100644 --- a/lib/rebaser-server/src/change_set_processor_task.rs +++ b/lib/rebaser-server/src/change_set_processor_task.rs @@ -545,8 +545,11 @@ mod handlers { use dal::{ ChangeSet, + ChangeSetId, + DalContext, Workspace, WorkspaceSnapshot, + WorkspaceSnapshotAddress, WorkspaceSnapshotError, WsEvent, action::{ @@ -554,6 +557,7 @@ mod handlers { ActionError, }, billing_publish, + change_set::ChangeSetResult, workspace_snapshot::{ DependentValueRoot, dependent_value_root::DependentValueRootError, @@ -580,14 +584,32 @@ mod handlers { use rebaser_core::api_types::{ ContentInfo, SerializeError, - enqueue_updates_request::EnqueueUpdatesRequest, + enqueue_updates_request::{ + ApplyToHeadRequestV4, + BeginApplyToHeadRequestV4, + ChangeSetStatusUpdateRequestV4, + CreateChangeSetRequestV4, + EnqueueUpdatesRequest, + EnqueueUpdatesRequestVCurrent, + RebaseRequestV4, + RequestId, + }, enqueue_updates_response::{ + ApplyToHeadStatus, + BeginApplyStatus, + CreateChangeSetStatus, EnqueueUpdatesResponse, EnqueueUpdatesResponseVCurrent, RebaseStatus, + UpdateChangeSetStatusStatus, }, }; - use si_data_nats::HeaderMap; + use si_data_nats::{ + HeaderMap, + NatsClient, + Subject, + }; + use si_events::ChangeSetStatus; use telemetry::prelude::*; use telemetry_nats::propagation; use telemetry_utils::metric; @@ -597,11 +619,20 @@ mod handlers { DEFAULT_ACTION_CONCURRENCY_LIMIT, app_state::AppState, }; - use crate::rebase::{ - RebaseError, - perform_rebase, - send_updates_to_edda_legacy_snapshot, - send_updates_to_edda_split_snapshot, + use crate::{ + apply::{ + ApplyError, + enqueue_apply_retry, + enqueue_apply_retry_with_delay, + get_apply_rebase_request, + mark_change_set_applied, + }, + rebase::{ + RebaseError, + perform_rebase, + send_updates_to_edda_legacy_snapshot, + send_updates_to_edda_split_snapshot, + }, }; #[remain::sorted] @@ -610,6 +641,9 @@ mod handlers { /// Failures related to Actions #[error("action error: {0}")] Action(#[from] ActionError), + /// Failures related to applying to head + #[error("apply to head error: {0}")] + ApplyToHead(#[from] crate::apply::ApplyError), /// Failures related to ChangeSets #[error("Change set error: {0}")] ChangeSet(#[from] dal::ChangeSetError), @@ -641,6 +675,12 @@ mod handlers { #[remain::sorted] #[derive(Debug, Error)] pub(crate) enum HandlerError { + #[error("message processing error during a rebase for an apply to HEAD, after commiting")] + ApplyInRebasePostCommit(#[source] HandlerErrorSource), + #[error("message processing error during a rebase for an apply to HEAD, before commiting")] + ApplyInRebasePreCommit(#[source] HandlerErrorSource), + #[error("message processing error after a rebase has succeeded during apply to HEAD")] + ApplyPostRebase(#[source] HandlerErrorSource), #[error("message processing error after commiting to change (delete message): {0}")] PostCommit(#[source] HandlerErrorSource), #[error("message processing error before commiting to change (retry message): {0}")] @@ -660,6 +700,10 @@ mod handlers { Self::PostCommit(err.into()) } + pub(crate) fn apply_post_rebase(err: impl Into) -> Self { + Self::ApplyPostRebase(err.into()) + } + pub(crate) fn transient_service(err: impl Into) -> Self { Self::TransientService(err.into()) } @@ -707,59 +751,366 @@ mod handlers { ); Response::default_bad_request() } + Self::ApplyPostRebase(err) => { + error!( + si.error.message = ?err, + "failed to finalize an apply after rebase", + ); + Response::default_internal_server_error() + } + Self::ApplyInRebasePostCommit(err) => { + error!( + si.error.message = ?err, + "failed to perform post-commit activities in rebase for apply to head", + ); + Response::default_internal_server_error() + } + Self::ApplyInRebasePreCommit(err) => { + error!( + si.error.message = ?err, + "failed to commit rebase for apply to head", + ); + Response::default_internal_server_error() + } } } } - pub(crate) async fn default( - State(state): State, - HeaderReply(maybe_reply): HeaderReply, - Negotiate(request): Negotiate, + async fn handle_begin_apply( + ctx: &mut DalContext, + state: AppState, + maybe_reply: Option, + request_id: RequestId, + request: BeginApplyToHeadRequestV4, ) -> Result<()> { let AppState { + nats, workspace_id, change_set_id, - nats, - edda, - ctx_builder, - run_notify, server_tracker, - features, + .. } = state; - let span = Span::current(); - span.record("si.workspace.id", workspace_id.to_string()); - span.record("si.change_set.id", change_set_id.to_string()); - let metric_label = format!("{workspace_id}:{change_set_id}"); - metric!(counter.rebaser.rebase_processing = 1, label = metric_label); + let mut apply_to_head_result = crate::apply::begin_apply(ctx, &request).await; - let mut ctx = ctx_builder - .build_for_change_set_as_system(workspace_id, change_set_id, None) + // Replays from HEAD could insert new DVU roots! So we need to keep the + // retrying going here as well + if let Err(ApplyError::DependentValueRootExists(previous_change_set_status)) = + &apply_to_head_result + { + debug!( + "DVU unfinished for change set {change_set_id}, retrying begin apply with delay" + ); + enqueue_apply_retry_with_delay( + ctx, + ApplyToHeadRequestV4 { + workspace_id: request.workspace_id, + head_change_set_id: request.head_change_set_id, + head_change_set_address: request.head_change_set_address, + change_set_to_apply_id: request.change_set_id, + event_session_id: request.event_session_id, + previous_change_set_status: (*previous_change_set_status).into(), + mode: Some(request.mode), + }, + request.head_change_set_address, + server_tracker, + ) + .await; + apply_to_head_result = Ok(*previous_change_set_status); + } + + if let Some(reply_subject) = maybe_reply { + let response = + EnqueueUpdatesResponse::new(EnqueueUpdatesResponseVCurrent::BeginApplyToHead { + id: request_id, + workspace_id, + change_set_id, + status: match &apply_to_head_result { + Ok(previous_change_set_status) + | Err(ApplyError::DependentValueRootExists(previous_change_set_status)) => { + BeginApplyStatus::Success { + previous_change_set_status: (*previous_change_set_status).into(), + } + } + Err(handler_error) => BeginApplyStatus::Error { + message: handler_error.to_string(), + }, + }, + }); + + publish_response(&nats, reply_subject, response).await?; + } + + apply_to_head_result + .map_err(HandlerError::post_commit) + .map(|_| ()) + } + + async fn publish_response( + nats: &NatsClient, + reply: Subject, + response: EnqueueUpdatesResponse, + ) -> Result<()> { + let mut info = ContentInfo::from(&response); + let (content_type, payload) = response.to_vec().map_err(Error::post_commit)?; + info.content_type = content_type.into(); + + let mut headers = HeaderMap::new(); + propagation::inject_headers(&mut headers); + info.inject_into_headers(&mut headers); + + nats.publish_with_headers(reply, headers, payload.into()) .await - .map_err(|err| match &err { - // - Failed to get connection from pool or other pg pool issue - dal::TransactionsError::PgPool(_) - // - Failed to determine change set via db query - | dal::TransactionsError::ChangeSet(_) - // - Failed to determine workspace via db query - // - Failed to determine snapshot address via db query - => HandlerError::transient_service(err), - dal::TransactionsError::Workspace(workspace_err) - => { - // - Failed to load snapshot from layer db after trying for 10s - // so we should not retry because if we gave it 10s, it'll never be there - if let dal::WorkspaceError::WorkspaceSnapshot(WorkspaceSnapshotError::WorkspaceSnapshotNotFetched) = workspace_err.as_ref() { - HandlerError::unprocessable(err) + .map_err(|err| Error::post_commit(ErrorSource::PublishReply(err)))?; + + Ok(()) + } + + async fn handle_apply_to_head( + ctx: &mut DalContext, + state: AppState, + maybe_reply: Option, + request_id: RequestId, + request: ApplyToHeadRequestV4, + ) -> Result<()> { + let workspace_id = request.workspace_id; + let change_set_to_apply_id = request.change_set_to_apply_id; + let head_change_set_id = request.head_change_set_id; + let nats = state.nats.clone(); + let server_tracker = state.server_tracker.clone(); + let previous_change_set_status = request.previous_change_set_status; + + enum ApplyResult { + Retry(WorkspaceSnapshotAddress), + RetryWithDelay(WorkspaceSnapshotAddress), + Success, + } + + // We want a subroutine here since we need to treat the entire operation + // as returning a single result, to simplify the response handling + async fn do_apply( + ctx: &mut DalContext, + change_set_to_apply_id: ChangeSetId, + head_change_set_id: ChangeSetId, + state: AppState, + request_id: RequestId, + request: ApplyToHeadRequestV4, + ) -> Result { + info!("applying change set {change_set_to_apply_id} to head {head_change_set_id}"); + let head_change_set = ChangeSet::get_by_id(ctx, head_change_set_id) + .await + .map_err(HandlerError::pre_commit)?; + + let head_change_set_address = head_change_set.workspace_snapshot_address; + // If the head change set has moved forward since the request was + // made, that means changes from the head change set have been, or + // are in the process of, being replayed onto the change set that is + // being applied. These changes might not yet be finished, so we + // should enqueue a retry message back onto the request stream for + // the change set that is being applied. This request will land + // *after* any replay messages, and so the apply message will be + // placed back into the HEAD change set *after* the replay messages + // have been processed and the change set is fully up to date. This + // process will continue until HEAD is "at rest" (that is, it has + // not changed in between the time the apply request was made and + // the apply message is handled here). + if head_change_set_address != request.head_change_set_address { + debug!( + "retrying apply of {change_set_to_apply_id} to {head_change_set_id} because head change set has moved forward" + ); + return Ok(ApplyResult::Retry(head_change_set_address)); + } + + // Also ensure we have not raced the DVU, since we don't want to + // apply until all the values have been computed. + if DependentValueRoot::roots_exist(ctx) + .await + .map_err(HandlerError::pre_commit)? + { + debug!( + "retrying apply of {change_set_to_apply_id} to {head_change_set_id} because head change set has not processed DVU roots" + ); + return Ok(ApplyResult::RetryWithDelay(head_change_set_address)); + } + + let apply_to_head_rebase_request = get_apply_rebase_request(ctx, request) + .await + .map_err(HandlerError::pre_commit)?; + + let rebase_result = if let Some(rebase_request) = apply_to_head_rebase_request { + // *do not* send the reply in handle_rebase, we will send our + // own reply below + handle_rebase(ctx, state, None, request_id, rebase_request) + .await + .map_err(|err| match err { + HandlerError::PostCommit(handler_error_source) => { + HandlerError::ApplyInRebasePostCommit(handler_error_source) + } + HandlerError::PreCommit(handler_error_source) + | HandlerError::Unprocessable(handler_error_source) => { + HandlerError::ApplyInRebasePreCommit(handler_error_source) + } + _ => err, + }) + } else { + Ok(()) + }; + + // We need to mark the change set as applied if the rebase was + // committed, even if there were post-commit failures. + match rebase_result { + Ok(()) => { + mark_change_set_applied(ctx, change_set_to_apply_id, head_change_set_id) + .await + .map_err(HandlerError::apply_post_rebase)?; + + Ok(ApplyResult::Success) + } + Err(HandlerError::ApplyInRebasePostCommit(_)) => { + if let Err(err) = + mark_change_set_applied(ctx, change_set_to_apply_id, head_change_set_id) + .await + { + error!( + si.error.message = ?err, + "failed to mark change set as applied after apply rebase post-commit failure" + ); } - else { - HandlerError::transient_service(err) + + // This should never be the Ok variant but we need this for types + rebase_result.map(|_| ApplyResult::Success) + } + // This should also never be the Ok variant but we need this for types + err => err.map(|_| ApplyResult::Success), + } + } + + async fn reset_status( + ctx: &DalContext, + change_set_id: ChangeSetId, + previous_change_set_status: ChangeSetStatus, + ) -> ChangeSetResult<()> { + let mut change_set = ChangeSet::get_by_id(ctx, change_set_id).await?; + + // Only reset if we are in ApplyStarted + if change_set.status != dal::ChangeSetStatus::ApplyStarted { + return Ok(()); + } + + change_set + .update_status(ctx, previous_change_set_status.into()) + .await?; + ctx.commit_no_rebase().await?; + Ok(()) + } + + let apply_result = do_apply( + ctx, + change_set_to_apply_id, + head_change_set_id, + state, + request_id, + request.clone(), + ) + .await; + + match &apply_result { + Err(apply_err) => match apply_err { + HandlerError::ApplyPostRebase(_) | HandlerError::ApplyInRebasePostCommit(_) => { + if let Err(err) = + mark_change_set_applied(ctx, change_set_to_apply_id, head_change_set_id) + .await + { + error!( + "Final fail safe attempt to mark change set after applied failed but committed: {}", + err + ); } } - _ => HandlerError::pre_commit(err) - })?; + // If we failed before the rebase was committed, we need to move + // the change set status back to what it was, otherwise we won't + // be able to retry the apply + _ => { + if let Err(err) = + reset_status(ctx, change_set_to_apply_id, previous_change_set_status).await + { + error!( + "Failed to reset change set {change_set_to_apply_id} status to {previous_change_set_status} after apply failure: {}", + err + ); + } + } + }, + Ok(ApplyResult::Retry(head_snapshot_address)) => { + // Re-enqueue the apply request with the current head change set address, + // re-enqueues will continue until HEAD is at rest. + info!("enqueuing retry for change set {change_set_to_apply_id}"); + if let Err(err) = enqueue_apply_retry(ctx, request, *head_snapshot_address).await { + error!( + "Failed to enqueue apply retry for change set {change_set_to_apply_id}: {}", + err + ); + } + } + Ok(ApplyResult::RetryWithDelay(head_snapshot_address)) => { + info!("enqueuing retry with delay for change set {change_set_to_apply_id}"); + enqueue_apply_retry_with_delay( + ctx, + request, + *head_snapshot_address, + server_tracker, + ) + .await; + } + Ok(ApplyResult::Success) => {} + } + + if let Some(reply_subject) = maybe_reply { + let response = + EnqueueUpdatesResponse::new(EnqueueUpdatesResponseVCurrent::ApplyToHead { + id: request_id, + workspace_id, + change_set_id: head_change_set_id, + status: match &apply_result { + Ok(ApplyResult::Success) => ApplyToHeadStatus::Success, + Ok(ApplyResult::Retry(_) | ApplyResult::RetryWithDelay(_)) => { + ApplyToHeadStatus::Retrying + } + Err(handler_error) => ApplyToHeadStatus::Error { + message: handler_error.to_string(), + }, + }, + }); + + publish_response(&nats, reply_subject, response).await?; + } + + apply_result.map(|_| ()) + } + + async fn handle_rebase( + ctx: &mut DalContext, + state: AppState, + maybe_reply: Option, + request_id: RequestId, + request: RebaseRequestV4, + ) -> Result<()> { + let AppState { + workspace_id, + change_set_id, + nats, + edda, + run_notify, + server_tracker, + features, + .. + } = state; + + let metric_label = format!("{workspace_id}:{change_set_id}"); + metric!(counter.rebaser.rebase_processing = 1, label = metric_label); let rebase_batch_address_kind_result = - match perform_rebase(&mut ctx, &edda, &request, &server_tracker, features).await { + match perform_rebase(ctx, &edda, &request, &server_tracker, features).await { Ok(rebase_batch_address_kind) => Ok(rebase_batch_address_kind), Err(rebase_error) => match &rebase_error { // - Failed to compute correct transforms. Failures here are related to the @@ -792,7 +1143,7 @@ mod handlers { let maybe_post_rebase_activities_result = if rebase_batch_address_kind_result.is_ok() { Some( - post_rebase_activities(workspace_id, change_set_id, &edda, run_notify, &mut ctx) + post_rebase_activities(workspace_id, change_set_id, &edda, run_notify, ctx) .await .map_err(HandlerError::post_commit), ) @@ -801,11 +1152,11 @@ mod handlers { }; // If a reply was requested, send it - if let Some(reply) = maybe_reply { - let response = EnqueueUpdatesResponse::new(EnqueueUpdatesResponseVCurrent { - id: request.id, - workspace_id: request.workspace_id, - change_set_id: request.change_set_id, + if let Some(reply_subject) = maybe_reply { + let response = EnqueueUpdatesResponse::new(EnqueueUpdatesResponseVCurrent::Rebase { + id: request_id, + workspace_id, + change_set_id, status: match &rebase_batch_address_kind_result { Ok(rebase_batch_address_kind) => RebaseStatus::Success { updates_performed: *rebase_batch_address_kind, @@ -816,28 +1167,18 @@ mod handlers { }, }); - let mut info = ContentInfo::from(&response); - let (content_type, payload) = response.to_vec().map_err(Error::post_commit)?; - info.content_type = content_type.into(); - - let mut headers = HeaderMap::new(); - propagation::inject_headers(&mut headers); - info.inject_into_headers(&mut headers); - - nats.publish_with_headers(reply, headers, payload.into()) - .await - .map_err(|err| Error::post_commit(ErrorSource::PublishReply(err)))?; + publish_response(&nats, reply_subject, response).await?; } // TODO(fnichol): hrm, is this *really* true that we've written to the change set. I mean, // yes but until a dvu has finished this is an incomplete view? - let mut event = WsEvent::change_set_written(&ctx, change_set_id) + let mut event = WsEvent::change_set_written(ctx, change_set_id) .await .map_err(Error::post_commit)?; event.set_workspace_pk(workspace_id); event.set_change_set_id(Some(change_set_id)); event - .publish_immediately(&ctx) + .publish_immediately(ctx) .await .map_err(Error::post_commit)?; @@ -856,6 +1197,203 @@ mod handlers { } } + // Create a new change set. This should be executed via the request stream + // for the HEAD change set, so that new change sets are perfectly in sync + // with the latest updates to HEAD, and so that replays from HEAD will not + // be able to race change set creation (since they must happen before or + // after the rebaser handles the latest update to HEAD). + async fn handle_create_change_set( + ctx: &DalContext, + state: AppState, + maybe_reply: Option, + request_id: RequestId, + request: CreateChangeSetRequestV4, + ) -> Result<()> { + let name = request.name; + + let AppState { + workspace_id, + change_set_id, + nats, + .. + } = state; + + let metric_label = format!("{workspace_id}:{change_set_id}"); + metric!(counter.rebaser.create_change_set = 1, label = metric_label); + + async fn fork_head(ctx: &DalContext, name: &str) -> Result { + let new_change_set = ChangeSet::fork_head(ctx, name) + .await + .map_err(HandlerError::pre_commit)?; + ctx.commit_no_rebase() + .await + .map_err(HandlerError::pre_commit)?; + + Ok(new_change_set.id) + } + + let new_change_set_result = fork_head(ctx, &name).await; + + if let Some(reply_subject) = maybe_reply { + let response = + EnqueueUpdatesResponse::new(EnqueueUpdatesResponseVCurrent::CreateChangeSet { + id: request_id, + workspace_id, + status: match &new_change_set_result { + Ok(new_change_set_id) => CreateChangeSetStatus::Success { + new_change_set_id: *new_change_set_id, + }, + Err(handler_error) => CreateChangeSetStatus::Error { + message: handler_error.to_string(), + }, + }, + }); + + publish_response(&nats, reply_subject, response).await?; + } + + metric!(counter.rebaser.create_change_set = -1, label = metric_label); + + new_change_set_result.map(|_| ()) + } + + // Move the change set's status on the change_set_pointers_table. This + // should happen in the request stream for the change set, to ensure that + // change set status updates happen before or after a new set of updates + // land on the change set. This will mean that concurrent updates that + // happen after the change set status is closed will be rejected. + async fn handle_change_set_status_update( + ctx: &DalContext, + state: AppState, + maybe_reply: Option, + request_id: RequestId, + request: ChangeSetStatusUpdateRequestV4, + ) -> Result<()> { + let AppState { + workspace_id, + change_set_id, + nats, + .. + } = state; + + let metric_label = format!("{workspace_id}:{change_set_id}"); + metric!( + counter.rebaser.change_set_status_update = 1, + label = metric_label + ); + + async fn update_status( + ctx: &DalContext, + change_set_id: ChangeSetId, + new_status: ChangeSetStatus, + ) -> Result { + let mut change_set = ChangeSet::get_by_id(ctx, change_set_id) + .await + .map_err(HandlerError::pre_commit)?; + let previous_status = change_set.status; + change_set + .update_status(ctx, new_status.into()) + .await + .map_err(HandlerError::pre_commit)?; + ctx.commit_no_rebase() + .await + .map_err(HandlerError::pre_commit)?; + + Ok(previous_status.into()) + } + + let update_status_result = update_status(ctx, change_set_id, request.status).await; + + if let Some(reply_subject) = maybe_reply { + let response = EnqueueUpdatesResponse::new( + EnqueueUpdatesResponseVCurrent::UpdateChangeSetStatus { + id: request_id, + workspace_id, + change_set_id, + status: match &update_status_result { + Ok(prev) => UpdateChangeSetStatusStatus::Success { + previous_change_set_status: *prev, + }, + Err(handler_error) => UpdateChangeSetStatusStatus::Error { + message: handler_error.to_string(), + }, + }, + }, + ); + + publish_response(&nats, reply_subject, response).await?; + } + + metric!( + counter.rebaser.change_set_status_update = -1, + label = metric_label + ); + + update_status_result.map(|_| ()) + } + + pub(crate) async fn default( + State(state): State, + HeaderReply(maybe_reply): HeaderReply, + Negotiate(request): Negotiate, + ) -> Result<()> { + let AppState { + workspace_id, + change_set_id, + ctx_builder, + .. + } = &state; + let span = Span::current(); + + span.record("si.workspace.id", workspace_id.to_string()); + span.record("si.change_set.id", change_set_id.to_string()); + + let mut ctx = ctx_builder + .build_for_change_set_as_system(*workspace_id, *change_set_id, None) + .await + .map_err(|err| match &err { + // - Failed to get connection from pool or other pg pool issue + dal::TransactionsError::PgPool(_) + // - Failed to determine change set via db query + | dal::TransactionsError::ChangeSet(_) + // - Failed to determine workspace via db query + // - Failed to determine snapshot address via db query + => HandlerError::transient_service(err), + dal::TransactionsError::Workspace(workspace_err) + => { + // - Failed to load snapshot from layer db after trying for 10s + // so we should not retry because if we gave it 10s, it'll never be there + if let dal::WorkspaceError::WorkspaceSnapshot(WorkspaceSnapshotError::WorkspaceSnapshotNotFetched) = workspace_err.as_ref() { + HandlerError::unprocessable(err) + } + else { + HandlerError::transient_service(err) + } + } + _ => HandlerError::pre_commit(err) + })?; + + let EnqueueUpdatesRequest::V4(request) = request; + match request { + EnqueueUpdatesRequestVCurrent::Rebase { + id, + request: rebase_request, + } => handle_rebase(&mut ctx, state, maybe_reply, id, rebase_request).await, + EnqueueUpdatesRequestVCurrent::BeginApplyToHead { id, request } => { + handle_begin_apply(&mut ctx, state, maybe_reply, id, request).await + } + EnqueueUpdatesRequestVCurrent::ApplyToHead { id, request } => { + handle_apply_to_head(&mut ctx, state, maybe_reply, id, request).await + } + EnqueueUpdatesRequestVCurrent::CreateChangeSet { id, request } => { + handle_create_change_set(&ctx, state, maybe_reply, id, request).await + } + EnqueueUpdatesRequestVCurrent::ChangeSetStatusUpdate { id, request } => { + handle_change_set_status_update(&ctx, state, maybe_reply, id, request).await + } + } + } + #[instrument( name = "rebase.post_rebase_activities", level = "info", @@ -872,7 +1410,7 @@ mod handlers { ))] async fn post_rebase_activities( workspace_id: dal::WorkspacePk, - change_set_id: dal::ChangeSetId, + change_set_id: ChangeSetId, edda: &EddaClient, run_notify: std::sync::Arc, ctx: &mut dal::DalContext, diff --git a/lib/rebaser-server/src/lib.rs b/lib/rebaser-server/src/lib.rs index 69a4469fb0..0b6e60299b 100644 --- a/lib/rebaser-server/src/lib.rs +++ b/lib/rebaser-server/src/lib.rs @@ -30,6 +30,7 @@ use thiserror::Error; mod app_state; +mod apply; mod change_set_processor_task; mod config; mod handlers; diff --git a/lib/rebaser-server/src/rebase.rs b/lib/rebaser-server/src/rebase.rs index 93455aa77b..556b0eb628 100644 --- a/lib/rebaser-server/src/rebase.rs +++ b/lib/rebaser-server/src/rebase.rs @@ -23,7 +23,7 @@ use dal::{ }; use edda_client::EddaClient; use pending_events::PendingEventsError; -use rebaser_core::api_types::enqueue_updates_request::EnqueueUpdatesRequest; +use rebaser_core::api_types::enqueue_updates_request::RebaseRequestV4; use shuttle_server::ShuttleError; use si_events::{ RebaseBatchAddressKind, @@ -98,7 +98,7 @@ type RebaseResult = Result; pub(crate) async fn perform_rebase( ctx: &mut DalContext, edda: &EddaClient, - request: &EnqueueUpdatesRequest, + request: &RebaseRequestV4, server_tracker: &TaskTracker, features: Features, ) -> RebaseResult { @@ -106,12 +106,11 @@ pub(crate) async fn perform_rebase( let workspace = get_workspace(ctx).await?; let updating_head = request.change_set_id == workspace.default_change_set_id(); - // Gather everything we need to detect conflicts and updates from the inbound message. let mut to_rebase_change_set = ChangeSet::get_by_id(ctx, request.change_set_id).await?; // if the change set isn't active, do not do this work - if !to_rebase_change_set.status.is_active() { - debug!("Attempted to rebase for abandoned change set. Early returning"); + if !to_rebase_change_set.status.is_active_or_applying() { + info!("Attempted to rebase for inactive change set. Early returning"); return Err(RebaseError::AbandonedChangeSet(to_rebase_change_set.id)); } @@ -159,11 +158,8 @@ pub(crate) async fn perform_rebase( } if updating_head && *workspace.pk() != WorkspacePk::NONE { - //todo(brit): what do we want to do about change sets that haven't - // been applied yet, but are approved? (like gh merge-queue) - // should we 'unapprove' them? - let all_open_change_sets = ChangeSet::list_active(ctx).await?; - for target_change_set in all_open_change_sets.into_iter().filter(|cs| { + let replayable_change_sets = ChangeSet::list_replayable(ctx).await?; + for target_change_set in replayable_change_sets.into_iter().filter(|cs| { cs.id != workspace.default_change_set_id() && cs.id != to_rebase_change_set.id && request.from_change_set_id != Some(cs.id) @@ -173,7 +169,7 @@ pub(crate) async fn perform_rebase( { let ctx_clone = ctx.clone(); server_tracker.spawn(async move { - debug!( + warn!( "replaying batch {} onto {} from {}", updates_address, target_change_set.id, to_rebase_change_set.id ); @@ -234,7 +230,7 @@ async fn rebase_split( ctx: &mut DalContext, to_rebase_workspace_snapshot_address: WorkspaceSnapshotAddress, edda: &EddaClient, - request: &EnqueueUpdatesRequest, + request: &RebaseRequestV4, _features: Features, updating_head: bool, to_rebase_change_set: &mut ChangeSet, @@ -328,7 +324,7 @@ async fn rebase_legacy( ctx: &mut DalContext, to_rebase_workspace_snapshot_address: WorkspaceSnapshotAddress, edda: &EddaClient, - request: &EnqueueUpdatesRequest, + request: &RebaseRequestV4, _features: Features, updating_head: bool, to_rebase_change_set: &mut ChangeSet, diff --git a/lib/sdf-core/src/dal_wrapper/change_set.rs b/lib/sdf-core/src/dal_wrapper/change_set.rs index da2acb2536..c5552f454f 100644 --- a/lib/sdf-core/src/dal_wrapper/change_set.rs +++ b/lib/sdf-core/src/dal_wrapper/change_set.rs @@ -97,14 +97,9 @@ pub async fn protected_apply_to_base_change_set( } } - // With no unsatisfied requirements, we can prepare to apply. - ChangeSet::prepare_for_apply_without_status_check(ctx).await?; - - // We need to commit our preparations before performing the apply. - ctx.commit().await?; - - // With the requirement check and preparations committed, we can finally perform the apply. - ChangeSet::apply_to_base_change_set(ctx).await?; + // With the requirement check satisfied, we can finally perform the apply. + // We skip the "status" check since it is performed above + ChangeSet::begin_apply_without_status_check(ctx).await?; Ok(()) } diff --git a/lib/sdf-server/src/garbage_collection.rs b/lib/sdf-server/src/garbage_collection.rs index 1193092dbc..a14397c384 100644 --- a/lib/sdf-server/src/garbage_collection.rs +++ b/lib/sdf-server/src/garbage_collection.rs @@ -86,7 +86,7 @@ impl SnapshotGarbageCollector { // Gather the WorkspaceSnapshotAddress of all open change sets. let open_statuses: Vec = ChangeSetStatus::iter() .filter_map(|status| { - if status.is_active() { + if status.is_active_or_applying() { Some(status.to_string()) } else { None diff --git a/lib/sdf-server/src/service/v2/change_set/apply.rs b/lib/sdf-server/src/service/v2/change_set/apply.rs index cfd16ac35d..164cfb8bfd 100644 --- a/lib/sdf-server/src/service/v2/change_set/apply.rs +++ b/lib/sdf-server/src/service/v2/change_set/apply.rs @@ -86,7 +86,7 @@ pub async fn apply( post_to_webhook(&ctx, workspace_pk, message.as_str()).await?; // WS Event fires from the dal - ctx.commit().await?; + ctx.commit_no_rebase().await?; } Ok(Json(action_list)) diff --git a/lib/sdf-server/src/service/v2/change_set/create_initialize_apply.rs b/lib/sdf-server/src/service/v2/change_set/create_initialize_apply.rs index fc8d814a0c..375645ff37 100644 --- a/lib/sdf-server/src/service/v2/change_set/create_initialize_apply.rs +++ b/lib/sdf-server/src/service/v2/change_set/create_initialize_apply.rs @@ -79,8 +79,7 @@ pub async fn create_initialize_apply( ) .await?; - let mut initialization_changeset_ctx = - ctx.clone_with_new_visibility(initialization_change_set.id.into()); + let mut initialization_changeset_ctx = ctx.clone_with_change_set(initialization_change_set.id); let initialization_result = initialize_and_apply(&mut initialization_changeset_ctx, request).await; @@ -315,7 +314,7 @@ async fn initialize_and_apply(ctx: &mut DalContext, request: Request) -> Result< let change_set = ctx.change_set()?.clone(); let old_status = change_set.status; - ChangeSet::prepare_for_force_apply(ctx).await?; + ChangeSet::force_change_set_approval(ctx).await?; ctx.write_audit_log( AuditLogKind::ApproveChangeSetApply { from_status: old_status.into(), @@ -324,7 +323,7 @@ async fn initialize_and_apply(ctx: &mut DalContext, request: Request) -> Result< ) .await?; - ChangeSet::apply_to_base_change_set(ctx).await?; + ChangeSet::begin_apply(ctx).await?; ctx.write_audit_log(AuditLogKind::ApplyChangeSet, change_set.name.clone()) .await?; diff --git a/lib/sdf-server/src/service/v2/change_set/force_apply.rs b/lib/sdf-server/src/service/v2/change_set/force_apply.rs index 18df1b3f6d..fd87c41fe4 100644 --- a/lib/sdf-server/src/service/v2/change_set/force_apply.rs +++ b/lib/sdf-server/src/service/v2/change_set/force_apply.rs @@ -28,12 +28,12 @@ pub async fn force_apply( Host(host_name): Host, Path((_workspace_pk, change_set_id)): Path<(WorkspacePk, ChangeSetId)>, ) -> Result<()> { - let mut ctx = builder + let ctx = builder .build(request_ctx.build(change_set_id.into())) .await?; let change_set = ChangeSet::get_by_id(&ctx, change_set_id).await?; let old_status = change_set.status; - ChangeSet::prepare_for_force_apply(&ctx).await?; + ChangeSet::force_change_set_approval(&ctx).await?; ctx.write_audit_log( AuditLogKind::ApproveChangeSetApply { from_status: old_status.into(), @@ -44,7 +44,7 @@ pub async fn force_apply( // We need to run a commit before apply so changes get saved ctx.commit().await?; - ChangeSet::apply_to_base_change_set(&mut ctx).await?; + ChangeSet::begin_apply(&ctx).await?; track( &posthog_client, @@ -57,7 +57,7 @@ pub async fn force_apply( }), ); - let change_set = ChangeSet::get_by_id(&ctx, ctx.visibility().change_set_id).await?; + let change_set = ChangeSet::get_by_id(&ctx, change_set_id).await?; ctx.write_audit_log(AuditLogKind::ApplyChangeSet, change_set.name) .await?; diff --git a/lib/sdf-test/src/helpers.rs b/lib/sdf-test/src/helpers.rs index ce614ff921..dedaa61301 100644 --- a/lib/sdf-test/src/helpers.rs +++ b/lib/sdf-test/src/helpers.rs @@ -1,9 +1,6 @@ //! This module provides [`SdfChangeSetTestHelpers`]. -use dal::{ - ChangeSet, - DalContext, -}; +use dal::DalContext; use dal_test::prelude::*; use sdf_core::dal_wrapper; @@ -23,14 +20,7 @@ impl SdfTestHelpers { dal_wrapper::change_set::approval_requirements_are_satisfied_or_error(ctx, spicedb_client) .await?; - // We need to prepare for apply, but use the new version that skips the status check. Why? - // Our new approval requirements check above replaces the old status check. - ChangeSet::prepare_for_apply_without_status_check(ctx).await?; - - // We can mostly follow the DAL test flow for applying a change set while factoring in - // approvals. The "mostly" part comes from the fact that we need to perform our own - // "prepare" step above rather than using the one from the DAL test helpers. - ChangeSetTestHelpers::apply_change_set_to_base_approvals_without_prepare_step(ctx).await?; + ChangeSetTestHelpers::apply_change_set_to_base(ctx).await?; Ok(()) } } diff --git a/lib/si-events-rs/src/change_set_status.rs b/lib/si-events-rs/src/change_set_status.rs index 36bbf5ec04..44ae87183c 100644 --- a/lib/si-events-rs/src/change_set_status.rs +++ b/lib/si-events-rs/src/change_set_status.rs @@ -15,6 +15,7 @@ use strum::{ pub enum ChangeSetStatus { Abandoned, Applied, + ApplyStarted, Approved, Failed, NeedsAbandonApproval,