System Initiative

use std::{ result, sync::Arc, }; use dal::ChangeSetError; use pinga_client::{ PingaClient, api_types::job_execution_response::JobExecutionResultVCurrent, }; use si_events::{ ChangeSetId, WorkspacePk, }; use telemetry::prelude::*; use telemetry_utils::metric; use thiserror::Error; use tokio::sync::Notify; use tokio_util::sync::CancellationToken; use crate::{ ServerMetadata, Shutdown, }; #[remain::sorted] #[derive(Debug, Error)] pub(crate) enum SerialDvuTaskError { #[error("change set error: {0}")] ChangeSet(#[from] ChangeSetError), /// Error when using a Pinga client #[error("pinga client error: {0}")] PingaClient(#[from] pinga_client::ClientError), /// When failing to do an operation using the [`WorkspaceSnapshot`] #[error("workspace snapshot error: {0}")] WorkspaceSnapshot(#[from] dal::WorkspaceSnapshotError), } type Result<T> = result::Result<T, SerialDvuTaskError>; pub(crate) struct SerialDvuTask { metadata: Arc<ServerMetadata>, workspace_id: WorkspacePk, change_set_id: ChangeSetId, pinga: PingaClient, run_notify: Arc<Notify>, quiesced_notify: Arc<Notify>, quiesced_token: CancellationToken, token: CancellationToken, } impl SerialDvuTask { const NAME: &'static str = "rebaser_server::serial_dvu_task"; #[allow(clippy::too_many_arguments)] pub(crate) fn create( metadata: Arc<ServerMetadata>, workspace_id: WorkspacePk, change_set_id: ChangeSetId, pinga: PingaClient, run_notify: Arc<Notify>, quiesced_notify: Arc<Notify>, quiesced_token: CancellationToken, token: CancellationToken, ) -> Self { Self { metadata, workspace_id, change_set_id, pinga, run_notify, quiesced_notify, quiesced_token, token, } } pub(crate) async fn try_run(self) -> Result<Shutdown> { metric!(counter.serial_dvu_task.change_set_in_progress = 1); let shutdown_cause = loop { tokio::select! { biased; // Signal to run a DVU has fired _ = self.run_notify.notified() => { info!( task = Self::NAME, service.instance.id = self.metadata.instance_id(), si.workspace.id = %self.workspace_id, si.change_set.id = %self.change_set_id, "notified, preparing dvu run", ); self.run_dvu().await; } // Signal to shutdown from a quiet period has fired _ = self.quiesced_notify.notified() => { info!( task = Self::NAME, service.instance.id = self.metadata.instance_id(), si.workspace.id = %self.workspace_id, si.change_set.id = %self.change_set_id, "quiesced notified, starting to shut down", ); // Fire the quiesced_token so that the processing task immediately stops // processing additional requests self.quiesced_token.cancel(); break Shutdown::Quiesced; } // Cancellation token has fired, time to shut down _ = self.token.cancelled() => { info!( task = Self::NAME, service.instance.id = self.metadata.instance_id(), si.workspace.id = %self.workspace_id, si.change_set.id = %self.change_set_id, "received cancellation, shutting down", ); break Shutdown::Graceful; } } }; info!( task = Self::NAME, cause = ?shutdown_cause, service.instance.id = self.metadata.instance_id(), si.workspace.id = %self.workspace_id, si.change_set.id = %self.change_set_id, "shutdown complete", ); metric!(counter.serial_dvu_task.change_set_in_progress = -1); Ok(shutdown_cause) } #[instrument( name = "serial_dvu_task.run_dvu", level = "info", skip_all, fields( service.instance.id = self.metadata.instance_id(), si.change_set.id = %self.change_set_id, si.workspace.id = %self.workspace_id, ), )] async fn run_dvu(&self) { metric!(counter.serial_dvu_task.dvu_running = 1); if let Err(err) = self.try_run_dvu().await { error!( si.error.message = ?err, "error encountered when waiting on dvu job from pinga; continuing", ); } metric!(counter.serial_dvu_task.dvu_running = -1); } #[inline] async fn try_run_dvu(&self) -> Result<()> { let (_request_id, response_fut) = self .pinga .await_dependent_values_update_job(self.workspace_id, self.change_set_id, false) .await?; // TODO(fnichol): here's another spot to consider a timeout, otherwise this task will wait // indefinitely for the Pinga DVU job to complete. // // Future work may change this response future idea an instead check the job state from a // KV store which is being actively updated by a Pinga task. Something like that to avoid // the sender terminating and this receiver never getting a reply either way. match response_fut.await { Ok(response) => match &response.result { JobExecutionResultVCurrent::Ok => { trace!("pinga job reported a successful job execution"); } JobExecutionResultVCurrent::Err { message } => { warn!( job_execution_error_report = message, "pinga job reported an error during execution; continuing", ); } }, Err(client_error) => return Err(client_error.into()), } Ok(()) } }