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,