use crate::worker::{FatalError, Worker, WorkerContext, WorkerError, WorkerResult, WorkerResultOk};
use assert_matches::assert_matches;
use chrono::{DateTime, Utc};
use concepts::prefixed_ulid::{DeploymentId, RunId};
use concepts::storage::{
AppendEventsToExecution, AppendRequest, AppendResponseToExecution, DbErrorGeneric,
DbErrorWrite, DbExecutor, DbPool, ExecutionLog, LockedExecution,
};
use concepts::time::{ClockFn, Sleep};
use concepts::{
ComponentId, ComponentRetryConfig, ComponentType, FunctionMetadata, StrVariant,
SupportedFunctionReturnValue,
};
use concepts::{ExecutionFailureKind, JoinSetId};
use concepts::{ExecutionId, FunctionFqn, prefixed_ulid::ExecutorId};
use concepts::{
FinishedExecutionError,
storage::{ExecutionRequest, Version},
};
use std::{
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
time::Duration,
};
use tokio::task::{AbortHandle, JoinHandle};
use tracing::{Instrument, Level, Span, debug, error, info, info_span, instrument, trace, warn};
#[derive(Debug, Clone)]
pub struct ExecConfig {
pub lock_expiry: Duration,
pub tick_sleep: Duration,
pub batch_size: u32,
pub component_id: ComponentId,
pub task_limiter: Option<Arc<tokio::sync::Semaphore>>,
pub executor_id: ExecutorId,
pub retry_config: ComponentRetryConfig,
pub locking_strategy: LockingStrategy,
}
pub struct ExecTask {
worker: Arc<dyn Worker>,
pub config: ExecConfig,
clock_fn: Box<dyn ClockFn>, db_pool: Arc<dyn DbPool>,
locking_strategy_holder: LockingStrategyHolder,
worker_count_tx: tokio::sync::watch::Sender<usize>,
executor_close_watcher: tokio::sync::watch::Receiver<bool>,
}
#[derive(derive_more::Debug, Default)]
pub struct ExecutionProgress {
#[debug(skip)]
#[allow(dead_code)]
executions: Vec<(ExecutionId, JoinHandle<()>)>,
}
impl ExecutionProgress {
#[cfg(feature = "test")]
pub async fn wait_for_tasks(self) -> Vec<ExecutionId> {
let mut vec = Vec::new();
for (exe, join_handle) in self.executions {
vec.push(exe);
join_handle.await.unwrap();
}
vec
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkerType {
Activity,
Workflow,
}
#[derive(derive_more::Debug)]
pub struct ExecutorTaskHandle {
#[debug(skip)]
is_closing: Arc<AtomicBool>,
#[debug(skip)]
abort_handle: AbortHandle,
component_id: ComponentId,
executor_id: ExecutorId,
deployment_id: DeploymentId,
executor_closing_signal_sender: tokio::sync::watch::Sender<bool>,
worker_count_rx: tokio::sync::watch::Receiver<usize>,
}
impl ExecutorTaskHandle {
#[instrument(name = "executor.close", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id,
deployment_id = %self.deployment_id))]
pub async fn close(&self) {
trace!("Gracefully closing");
self.is_closing.store(true, Ordering::Relaxed);
while !self.abort_handle.is_finished() {
tokio::time::sleep(Duration::from_millis(1)).await;
}
trace!("Signaling workflow tasks to unlock");
let _ = self.executor_closing_signal_sender.send(true);
let mut worker_count_rx = self.worker_count_rx.clone();
loop {
tokio::select! {
() = tokio::time::sleep(Duration::from_secs(1)) => {
debug!("Waiting for {} workers to shut down", *self.worker_count_rx.borrow());
}
_ = worker_count_rx.wait_for(|&count| count == 0) => {
break;
}
}
}
debug!("Gracefully closed");
}
#[must_use]
pub fn component_id(&self) -> &ComponentId {
&self.component_id
}
}
impl Drop for ExecutorTaskHandle {
#[instrument(level = Level::DEBUG, name = "executor.drop", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id))]
fn drop(&mut self) {
if self.abort_handle.is_finished() {
return;
}
warn!("Aborting the executor task");
self.abort_handle.abort();
}
}
#[cfg(feature = "test")]
pub fn extract_exported_ffqns_noext_test(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
extract_exported_ffqns_noext(worker)
}
fn extract_exported_ffqns_noext(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
worker
.exported_functions_noext()
.iter()
.map(|FunctionMetadata { ffqn, .. }| ffqn.clone())
.collect::<Arc<_>>()
}
#[derive(Debug, Clone, Copy)]
pub enum LockingStrategy {
ByFfqns,
ByComponentDigest,
}
impl LockingStrategy {
fn holder(&self, ffqns: Arc<[FunctionFqn]>) -> LockingStrategyHolder {
match self {
LockingStrategy::ByFfqns => LockingStrategyHolder::ByFfqns(ffqns),
LockingStrategy::ByComponentDigest => LockingStrategyHolder::ByComponentId,
}
}
}
enum LockingStrategyHolder {
ByFfqns(Arc<[FunctionFqn]>),
ByComponentId,
}
impl ExecTask {
#[cfg(feature = "test")]
pub fn new_test(
config: ExecConfig,
worker: Arc<dyn Worker>,
clock_fn: Box<dyn ClockFn>,
db_pool: Arc<dyn DbPool>,
ffqns: Arc<[FunctionFqn]>,
) -> Self {
let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
ExecTask {
worker,
locking_strategy_holder: config.locking_strategy.holder(ffqns),
config,
clock_fn,
db_pool,
worker_count_tx,
executor_close_watcher: tokio::sync::watch::channel(false).1,
}
}
#[cfg(feature = "test")]
pub fn new_all_ffqns_test(
worker: Arc<dyn Worker>,
config: ExecConfig,
clock_fn: Box<dyn ClockFn>,
db_pool: Arc<dyn DbPool>,
) -> Self {
let ffqns = extract_exported_ffqns_noext(worker.as_ref());
let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
Self {
worker,
locking_strategy_holder: config.locking_strategy.holder(ffqns),
config,
clock_fn,
db_pool,
worker_count_tx,
executor_close_watcher: tokio::sync::watch::channel(false).1,
}
}
#[cfg(feature = "test")]
pub fn new_all_ffqns_test_with_close_watcher(
worker: Arc<dyn Worker>,
config: ExecConfig,
clock_fn: Box<dyn ClockFn>,
db_pool: Arc<dyn DbPool>,
executor_close_watcher: tokio::sync::watch::Receiver<bool>,
) -> Self {
let ffqns = extract_exported_ffqns_noext(worker.as_ref());
let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
Self {
worker,
locking_strategy_holder: config.locking_strategy.holder(ffqns),
config,
clock_fn,
db_pool,
worker_count_tx,
executor_close_watcher,
}
}
pub fn spawn_new(
deployment_id: DeploymentId,
worker: Arc<dyn Worker>,
config: ExecConfig,
clock_fn: Box<dyn ClockFn>,
db_pool: Arc<dyn DbPool>,
sleep: impl Sleep + Clone + 'static,
) -> ExecutorTaskHandle {
let is_closing = Arc::new(AtomicBool::default());
let is_closing_inner = is_closing.clone();
let ffqns = extract_exported_ffqns_noext(worker.as_ref());
let component_id = config.component_id.clone();
let executor_id = config.executor_id;
let (worker_count_tx, worker_count_rx) = tokio::sync::watch::channel(0);
let (executor_closing_signal_sender, executor_close_watcher) =
tokio::sync::watch::channel(false);
let abort_handle = tokio::spawn(async move {
debug!(executor_id = %config.executor_id, component_id = %config.component_id, "Spawned executor");
let lock_strategy_holder = config.locking_strategy.holder(ffqns);
let task = ExecTask {
worker,
config,
db_pool,
locking_strategy_holder: lock_strategy_holder,
clock_fn: clock_fn.clone_box(),
worker_count_tx,
executor_close_watcher,
};
let mut old_err = None;
while !is_closing_inner.load(Ordering::Relaxed) {
let res = task.db_pool.db_exec_conn().await;
let res = log_err_if_new(res, &mut old_err);
if let Ok(db_exec) = res {
let _ = task.tick(db_exec.as_ref(), clock_fn.now(), RunId::generate(), deployment_id).await;
db_exec
.wait_for_pending_by_component_digest(clock_fn.now(), &task.config.component_id.component_digest, {
let sleep = sleep.clone();
Box::pin(async move { sleep.sleep(task.config.tick_sleep).await })})
.await;
} else {
sleep.sleep(task.config.tick_sleep).await;
}
}
})
.abort_handle();
ExecutorTaskHandle {
is_closing,
abort_handle,
component_id,
executor_id,
deployment_id,
executor_closing_signal_sender,
worker_count_rx,
}
}
fn acquire_task_permits(&self) -> Vec<Option<tokio::sync::OwnedSemaphorePermit>> {
if let Some(task_limiter) = &self.config.task_limiter {
let mut locks = Vec::new();
for _ in 0..self.config.batch_size {
if let Ok(permit) = task_limiter.clone().try_acquire_owned() {
locks.push(Some(permit));
} else {
break;
}
}
locks
} else {
let mut vec = Vec::with_capacity(self.config.batch_size as usize);
for _ in 0..self.config.batch_size {
vec.push(None);
}
vec
}
}
#[cfg(feature = "test")]
pub async fn tick_test(&self, executed_at: DateTime<Utc>, run_id: RunId) -> ExecutionProgress {
use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
let db_exec = self.db_pool.db_exec_conn().await.unwrap();
self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
.await
.unwrap()
}
#[cfg(feature = "test")]
pub async fn tick_test_await(
&self,
executed_at: DateTime<Utc>,
run_id: RunId,
) -> Vec<ExecutionId> {
use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
let db_exec = self.db_pool.db_exec_conn().await.unwrap();
self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
.await
.unwrap()
.wait_for_tasks()
.await
}
#[instrument(level = Level::TRACE, name = "executor.tick" skip_all, fields(executor_id = %self.config.executor_id, component_id = %self.config.component_id))]
async fn tick(
&self,
db_exec: &dyn DbExecutor,
executed_at: DateTime<Utc>,
run_id: RunId,
deployment_id: DeploymentId,
) -> Result<ExecutionProgress, DbErrorWrite> {
let locked_executions = {
let mut permits = self.acquire_task_permits();
if permits.is_empty() {
return Ok(ExecutionProgress::default());
}
let lock_expires_at = executed_at + self.config.lock_expiry;
let batch_size = u32::try_from(permits.len()).expect("ExecConfig.batch_size is u32");
let locked_executions = match &self.locking_strategy_holder {
LockingStrategyHolder::ByFfqns(ffqns) => {
db_exec
.lock_pending_by_ffqns(
batch_size,
executed_at, ffqns.clone(),
executed_at, self.config.component_id.clone(),
deployment_id,
self.config.executor_id,
lock_expires_at,
run_id,
self.config.retry_config,
)
.await?
}
LockingStrategyHolder::ByComponentId => {
db_exec
.lock_pending_by_component_digest(
batch_size,
executed_at, &self.config.component_id,
deployment_id,
executed_at, self.config.executor_id,
lock_expires_at,
run_id,
self.config.retry_config,
)
.await?
}
};
while permits.len() > locked_executions.len() {
permits.pop();
}
assert_eq!(permits.len(), locked_executions.len());
locked_executions.into_iter().zip(permits)
};
let mut executions = Vec::with_capacity(locked_executions.len());
for (locked_execution, permit) in locked_executions {
let execution_id = locked_execution.execution_id.clone();
let join_handle = {
let worker = self.worker.clone();
let db_pool = self.db_pool.clone();
let clock_fn = self.clock_fn.clone_box();
let worker_span = info_span!(parent: None, "worker",
"otel.name" = format!("worker {}", locked_execution.ffqn),
%execution_id, %run_id,
ffqn = %locked_execution.ffqn,
executor_id = %self.config.executor_id,
component_id = %self.config.component_id,
%deployment_id,
);
locked_execution.metadata.enrich(&worker_span);
let component_type = self.config.component_id.component_type;
let worker_count_tx = self.worker_count_tx.clone();
worker_count_tx.send_modify(|n| *n += 1);
let executor_close_watcher = self.executor_close_watcher.clone();
tokio::spawn({
let worker_span2 = worker_span.clone();
let retry_config = self.config.retry_config;
async move {
let _permit = permit;
let res = Self::run_worker(
component_type,
worker,
db_pool.as_ref(),
clock_fn,
locked_execution,
retry_config,
worker_span2,
executor_close_watcher
)
.await;
if let Err(db_error) = res {
error!("Got db error `{db_error:?}`, expecting watcher to mark execution as timed out");
}
worker_count_tx.send_modify(|n| *n -= 1);
}
.instrument(worker_span)
})
};
executions.push((execution_id, join_handle));
}
Ok(ExecutionProgress { executions })
}
#[expect(clippy::too_many_arguments)]
async fn run_worker(
component_type: ComponentType,
worker: Arc<dyn Worker>,
db_pool: &dyn DbPool,
clock_fn: Box<dyn ClockFn>,
locked_execution: LockedExecution,
retry_config: ComponentRetryConfig,
worker_span: Span,
executor_close_watcher: tokio::sync::watch::Receiver<bool>,
) -> Result<(), DbErrorWrite> {
debug!("Worker::run starting");
trace!(
version = %locked_execution.next_version,
params = ?locked_execution.params,
event_history = ?locked_execution.event_history,
"Worker::run starting"
);
let can_be_retried = ExecutionLog::can_be_retried_after(
locked_execution.intermittent_event_count + 1,
retry_config.max_retries,
retry_config.retry_exp_backoff,
);
let unlock_expiry_on_limit_reached =
ExecutionLog::compute_retry_duration_when_retrying_forever(
locked_execution.intermittent_event_count + 1,
retry_config.retry_exp_backoff,
);
let ctx = WorkerContext {
execution_id: locked_execution.execution_id.clone(),
metadata: locked_execution.metadata,
ffqn: locked_execution.ffqn,
params: locked_execution.params,
event_history: locked_execution.event_history,
responses: locked_execution.responses,
version: locked_execution.next_version,
can_be_retried: can_be_retried.is_some(),
locked_event: locked_execution.locked_event,
worker_span,
executor_close_watcher,
};
let worker_result = worker.run(ctx).await;
debug!("Worker::run finished {worker_result:?}");
let result_obtained_at = clock_fn.now();
match Self::worker_result_to_execution_event(
component_type,
locked_execution.execution_id,
worker_result,
result_obtained_at,
locked_execution.parent,
can_be_retried,
unlock_expiry_on_limit_reached,
)? {
Some(append) => {
trace!("Appending {append:?}");
let db_exec = db_pool.db_exec_conn().await?;
match append {
AppendOrCancel::Cancel {
execution_id,
cancelled_at,
} => db_exec
.cancel_activity_with_retries(&execution_id, cancelled_at)
.await
.map(|_| ()),
AppendOrCancel::Other(append) => append.append(db_exec.as_ref()).await,
}
}
None => Ok(()),
}
}
fn worker_result_to_execution_event(
component_type: ComponentType,
execution_id: ExecutionId,
worker_result: WorkerResult,
result_obtained_at: DateTime<Utc>,
parent: Option<(ExecutionId, JoinSetId)>,
can_be_retried: Option<Duration>,
unlock_expiry_on_limit_reached: Duration,
) -> Result<Option<AppendOrCancel>, DbErrorWrite> {
Ok(match worker_result {
WorkerResult::Ok(WorkerResultOk::RunFinished {
retval: ref retval @ SupportedFunctionReturnValue::Err(ref result_err),
version,
http_client_traces,
}) if component_type == ComponentType::Activity
&& can_be_retried.is_some()
&& !retval.is_permanent_variant() =>
{
let detail = serde_json::to_string(result_err)
.expect("SupportedFunctionReturnValue should be serializable to JSON");
let duration = can_be_retried.expect(
"ActivityReturnedError must not be returned when retries are exhausted",
);
let expires_at = result_obtained_at + duration;
debug!("Retrying ActivityReturnedError after {duration:?} at {expires_at}");
let primary_event = ExecutionRequest::TemporarilyFailed {
backoff_expires_at: expires_at,
reason: StrVariant::Static("activity returned error"),
detail: Some(detail),
http_client_traces,
};
Some(AppendOrCancel::Other(Append {
created_at: result_obtained_at,
primary_event: AppendRequest {
created_at: result_obtained_at,
event: primary_event,
},
execution_id,
version,
child_finished: None,
}))
}
WorkerResult::Ok(WorkerResultOk::RunFinished {
retval: result,
version,
http_client_traces,
}) => {
info!("Execution finished: {result}");
let child_finished =
parent.map(
|(parent_execution_id, parent_join_set)| ChildFinishedResponse {
parent_execution_id,
parent_join_set,
result: result.clone(),
},
);
let primary_event = AppendRequest {
created_at: result_obtained_at,
event: ExecutionRequest::Finished {
retval: result,
http_client_traces,
},
};
Some(AppendOrCancel::Other(Append {
created_at: result_obtained_at,
primary_event,
execution_id,
version,
child_finished,
}))
}
WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher) => None,
WorkerResult::Err(err) => {
let reason_generic = err.to_string();
let (primary_event, child_finished, version) = match err {
WorkerError::ExecutorClosing(version) => {
let primary_event = ExecutionRequest::Unlocked {
backoff_expires_at: result_obtained_at, reason: "executor closing".into(),
};
(primary_event, None, version)
}
WorkerError::TemporaryTimeout {
http_client_traces,
version,
} => {
if let Some(duration) = can_be_retried {
let backoff_expires_at = result_obtained_at + duration;
info!(
"Temporary timeout, retrying after {duration:?} at {backoff_expires_at}"
);
(
ExecutionRequest::TemporarilyTimedOut {
backoff_expires_at,
http_client_traces,
},
None,
version,
)
} else {
info!("Execution timed out");
let result = SupportedFunctionReturnValue::ExecutionError(
FinishedExecutionError {
kind: ExecutionFailureKind::TimedOut,
reason: None,
detail: None,
},
);
let child_finished =
parent.map(|(parent_execution_id, parent_join_set)| {
ChildFinishedResponse {
parent_execution_id,
parent_join_set,
result: result.clone(),
}
});
(
ExecutionRequest::Finished {
retval: result,
http_client_traces,
},
child_finished,
version,
)
}
}
WorkerError::DbError(db_error) => {
return Err(db_error);
}
WorkerError::ActivityTrap {
reason: _, trap_kind,
detail,
version,
http_client_traces,
} => {
if let Some(duration) = can_be_retried {
let expires_at = result_obtained_at + duration;
debug!(
"Retrying activity with `{trap_kind}` execution after {duration:?} at {expires_at}"
);
(
ExecutionRequest::TemporarilyFailed {
reason: StrVariant::from(reason_generic),
backoff_expires_at: expires_at,
detail,
http_client_traces,
},
None,
version,
)
} else {
info!(
"Activity with `{trap_kind}` marked as permanent failure - {reason_generic}"
);
let result = SupportedFunctionReturnValue::ExecutionError(
FinishedExecutionError {
reason: Some(reason_generic),
kind: ExecutionFailureKind::Uncategorized,
detail,
},
);
let child_finished =
parent.map(|(parent_execution_id, parent_join_set)| {
ChildFinishedResponse {
parent_execution_id,
parent_join_set,
result: result.clone(),
}
});
(
ExecutionRequest::Finished {
retval: result,
http_client_traces,
},
child_finished,
version,
)
}
}
WorkerError::ActivityPreopenedDirError {
reason,
detail,
version,
} => {
let http_client_traces = None;
if let Some(duration) = can_be_retried {
let expires_at = result_obtained_at + duration;
debug!(
"Retrying activity with ActivityPreopenedDirError `{reason}` execution after {duration:?} at {expires_at}"
);
(
ExecutionRequest::TemporarilyFailed {
reason: StrVariant::from(reason_generic),
backoff_expires_at: expires_at,
detail: Some(detail),
http_client_traces,
},
None,
version,
)
} else {
info!(
"Activity with ActivityPreopenedDirError `{reason}` marked as permanent failure - {reason_generic}"
);
let result = SupportedFunctionReturnValue::ExecutionError(
FinishedExecutionError {
reason: Some(reason_generic),
kind: ExecutionFailureKind::Uncategorized,
detail: Some(detail),
},
);
let child_finished =
parent.map(|(parent_execution_id, parent_join_set)| {
ChildFinishedResponse {
parent_execution_id,
parent_join_set,
result: result.clone(),
}
});
(
ExecutionRequest::Finished {
retval: result,
http_client_traces,
},
child_finished,
version,
)
}
}
WorkerError::LimitReached {
reason,
version: new_version,
} => {
let expires_at = result_obtained_at + unlock_expiry_on_limit_reached;
warn!(
"Limit reached: {reason}, unlocking after {unlock_expiry_on_limit_reached:?} at {expires_at}"
);
(
ExecutionRequest::Unlocked {
backoff_expires_at: expires_at,
reason: StrVariant::from(reason),
},
None,
new_version,
)
}
WorkerError::FatalError(FatalError::Cancelled, _version) => {
return Ok(Some(AppendOrCancel::Cancel {
execution_id,
cancelled_at: result_obtained_at,
}));
}
WorkerError::FatalError(fatal_error, version) => {
warn!("Fatal worker error - {fatal_error:?}");
let result = SupportedFunctionReturnValue::ExecutionError(
FinishedExecutionError::from(fatal_error),
);
let child_finished =
parent.map(|(parent_execution_id, parent_join_set)| {
ChildFinishedResponse {
parent_execution_id,
parent_join_set,
result: result.clone(),
}
});
(
ExecutionRequest::Finished {
retval: result,
http_client_traces: None,
},
child_finished,
version,
)
}
};
Some(AppendOrCancel::Other(Append {
created_at: result_obtained_at,
primary_event: AppendRequest {
created_at: result_obtained_at,
event: primary_event,
},
execution_id,
version,
child_finished,
}))
}
})
}
}
#[derive(Debug, Clone)]
pub(crate) struct ChildFinishedResponse {
pub(crate) parent_execution_id: ExecutionId,
pub(crate) parent_join_set: JoinSetId,
pub(crate) result: SupportedFunctionReturnValue,
}
#[derive(Debug, Clone)]
#[expect(clippy::large_enum_variant)]
pub(crate) enum AppendOrCancel {
Cancel {
execution_id: ExecutionId,
cancelled_at: DateTime<Utc>,
},
Other(Append),
}
#[derive(Debug, Clone)]
pub(crate) struct Append {
pub(crate) created_at: DateTime<Utc>,
pub(crate) primary_event: AppendRequest,
pub(crate) execution_id: ExecutionId,
pub(crate) version: Version,
pub(crate) child_finished: Option<ChildFinishedResponse>,
}
impl Append {
pub(crate) async fn append(self, db_exec: &dyn DbExecutor) -> Result<(), DbErrorWrite> {
if let Some(child_finished) = self.child_finished {
assert_matches!(
&self.primary_event,
AppendRequest {
event: ExecutionRequest::Finished { .. },
..
}
);
let child_execution_id = assert_matches!(self.execution_id.clone(), ExecutionId::Derived(derived) => derived);
let events = AppendEventsToExecution {
execution_id: self.execution_id,
version: self.version.clone(),
batch: vec![self.primary_event],
};
let response = AppendResponseToExecution {
parent_execution_id: child_finished.parent_execution_id,
created_at: self.created_at,
join_set_id: child_finished.parent_join_set,
child_execution_id,
finished_version: self.version, result: child_finished.result,
};
db_exec
.append_batch_respond_to_parent(events, response, self.created_at)
.await?;
} else {
db_exec
.append(self.execution_id, self.version, self.primary_event)
.await?;
}
Ok(())
}
}
fn log_err_if_new<T>(
res: Result<T, DbErrorGeneric>,
old_err: &mut Option<DbErrorGeneric>,
) -> Result<T, ()> {
match (res, &old_err) {
(Ok(ok), _) => {
*old_err = None;
Ok(ok)
}
(Err(err), Some(old)) if err == *old => Err(()),
(Err(err), _) => {
warn!("Tick failed: {err:?}");
*old_err = Some(err);
Err(())
}
}
}
#[cfg(any(test, feature = "test"))]
pub mod simple_worker {
use crate::worker::{Worker, WorkerContext, WorkerResult};
use async_trait::async_trait;
use concepts::{
FunctionFqn, FunctionMetadata, ParameterTypes, RETURN_TYPE_DUMMY,
storage::{HistoryEvent, Version},
};
use indexmap::IndexMap;
use std::sync::Arc;
use tracing::trace;
pub(crate) const FFQN_SOME: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn");
pub type SimpleWorkerResultMap =
Arc<std::sync::Mutex<IndexMap<Version, (Vec<HistoryEvent>, WorkerResult)>>>;
#[derive(Clone, Debug)]
pub struct SimpleWorker {
pub worker_results_rev: SimpleWorkerResultMap,
pub ffqn: FunctionFqn,
exported: [FunctionMetadata; 1],
}
impl SimpleWorker {
#[must_use]
pub fn with_single_result(res: WorkerResult) -> Self {
Self::with_worker_results_rev(Arc::new(std::sync::Mutex::new(IndexMap::from([(
Version::new(2),
(vec![], res),
)]))))
}
#[must_use]
pub fn with_ffqn(self, ffqn: FunctionFqn) -> Self {
Self {
worker_results_rev: self.worker_results_rev,
exported: [FunctionMetadata {
ffqn: ffqn.clone(),
parameter_types: ParameterTypes::default(),
return_type: RETURN_TYPE_DUMMY,
extension: None,
submittable: true,
}],
ffqn,
}
}
#[must_use]
pub fn with_worker_results_rev(worker_results_rev: SimpleWorkerResultMap) -> Self {
Self {
worker_results_rev,
ffqn: FFQN_SOME,
exported: [FunctionMetadata {
ffqn: FFQN_SOME,
parameter_types: ParameterTypes::default(),
return_type: RETURN_TYPE_DUMMY,
extension: None,
submittable: true,
}],
}
}
}
#[async_trait]
impl Worker for SimpleWorker {
async fn run(&self, ctx: WorkerContext) -> WorkerResult {
let (expected_version, (expected_eh, worker_result)) =
self.worker_results_rev.lock().unwrap().pop().unwrap();
trace!(%expected_version, version = %ctx.version, ?expected_eh, eh = ?ctx.event_history, "Running SimpleWorker");
assert_eq!(expected_version, ctx.version);
assert_eq!(
expected_eh,
ctx.event_history
.iter()
.map(|(event, _version)| event.clone())
.collect::<Vec<_>>()
);
worker_result
}
fn exported_functions_noext(&self) -> &[FunctionMetadata] {
&self.exported
}
}
}
#[cfg(test)]
mod tests {
use self::simple_worker::SimpleWorker;
use super::*;
use crate::{expired_timers_watcher, worker::WorkerResult};
use assert_matches::assert_matches;
use async_trait::async_trait;
use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
use concepts::storage::{
CreateRequest, DbConnectionTest, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent,
};
use concepts::storage::{DbPoolCloseable, LockedBy};
use concepts::storage::{
ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState, PendingStatePendingAt,
};
use concepts::time::{ConstClock, Now};
use concepts::{
FunctionMetadata, JoinSetKind, ParameterTypes, Params, RETURN_TYPE_DUMMY,
SUPPORTED_RETURN_VALUE_OK_EMPTY, StrVariant, SupportedFunctionReturnValue, TrapKind,
};
use db_tests::Database;
use indexmap::IndexMap;
use rstest::rstest;
use simple_worker::FFQN_SOME;
use std::{fmt::Debug, future::Future, ops::Deref, sync::Arc};
use test_db_macro::expand_enum_database;
use test_utils::set_up;
use test_utils::sim_clock::SimClock;
pub(crate) const FFQN_CHILD: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn-child");
async fn tick_fn<W: Worker + Debug>(
config: ExecConfig,
clock_fn: Box<dyn ClockFn>,
db_pool: Arc<dyn DbPool>,
worker: Arc<W>,
executed_at: DateTime<Utc>,
) -> Vec<ExecutionId> {
trace!("Ticking with {worker:?}");
let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
let executor = ExecTask::new_test(config, worker, clock_fn, db_pool, ffqns);
executor
.tick_test_await(executed_at, RunId::generate())
.await
}
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn execute_simple_lifecycle_tick_based(
database: Database,
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
set_up();
let created_at = Now.now();
let (_guard, db_pool, db_close) = database.set_up().await;
let db_connection = db_pool.connection_test().await.unwrap();
execute_simple_lifecycle_tick_based_inner(
db_connection.as_ref(),
db_pool.clone(),
Box::new(ConstClock(created_at)),
locking_strategy,
)
.await;
drop(db_connection);
db_close.close().await;
}
async fn execute_simple_lifecycle_tick_based_inner(
db_connection: &dyn DbConnectionTest,
db_pool: Arc<dyn DbPool>,
clock_fn: Box<dyn ClockFn>,
locking_strategy: LockingStrategy,
) {
let created_at = clock_fn.now();
let exec_config = ExecConfig {
batch_size: 1,
lock_expiry: Duration::from_secs(1),
tick_sleep: Duration::from_millis(100),
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config: ComponentRetryConfig::ZERO,
locking_strategy,
};
let execution_log = create_and_tick(
CreateAndTickConfig {
execution_id: ExecutionId::generate(),
created_at,
executed_at: created_at,
},
clock_fn,
db_connection,
db_pool,
exec_config,
Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
WorkerResultOk::RunFinished {
retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
version: Version::new(2),
http_client_traces: None,
},
))),
tick_fn,
)
.await;
assert_matches!(
execution_log.events.get(2).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Finished {
retval: SupportedFunctionReturnValue::Ok(None),
http_client_traces: None
},
created_at: _,
backtrace_id: None,
version: Version(2),
}
);
}
#[rstest]
#[tokio::test]
async fn execute_simple_lifecycle_task_based_mem(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
set_up();
let created_at = Now.now();
let clock_fn = Box::new(ConstClock(created_at));
let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
let exec_config = ExecConfig {
batch_size: 1,
lock_expiry: Duration::from_secs(1),
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config: ComponentRetryConfig::ZERO,
locking_strategy,
};
let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
WorkerResultOk::RunFinished {
retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
version: Version::new(2),
http_client_traces: None,
},
)));
let db_connection = db_pool.connection_test().await.unwrap();
let execution_log = create_and_tick(
CreateAndTickConfig {
execution_id: ExecutionId::generate(),
created_at,
executed_at: created_at,
},
clock_fn,
db_connection.as_ref(),
db_pool,
exec_config,
worker,
tick_fn,
)
.await;
assert_matches!(
execution_log.events.get(2).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Finished {
retval: SupportedFunctionReturnValue::Ok(None),
http_client_traces: None
},
created_at: _,
backtrace_id: None,
version: Version(2),
}
);
db_close.close().await;
}
struct CreateAndTickConfig {
execution_id: ExecutionId,
created_at: DateTime<Utc>,
executed_at: DateTime<Utc>,
}
async fn create_and_tick<
W: Worker,
T: FnMut(ExecConfig, Box<dyn ClockFn>, Arc<dyn DbPool>, Arc<W>, DateTime<Utc>) -> F,
F: Future<Output = Vec<ExecutionId>>,
>(
config: CreateAndTickConfig,
clock_fn: Box<dyn ClockFn>,
db_connection: &dyn DbConnectionTest,
db_pool: Arc<dyn DbPool>,
exec_config: ExecConfig,
worker: Arc<W>,
mut tick: T,
) -> ExecutionLog {
db_connection
.create(CreateRequest {
created_at: config.created_at,
execution_id: config.execution_id.clone(),
ffqn: FFQN_SOME,
params: Params::empty(),
parent: None,
metadata: concepts::ExecutionMetadata::empty(),
scheduled_at: config.created_at,
component_id: ComponentId::dummy_activity(),
deployment_id: DEPLOYMENT_ID_DUMMY,
scheduled_by: None,
})
.await
.unwrap();
tick(exec_config, clock_fn, db_pool, worker, config.executed_at).await;
let execution_log = db_connection.get(&config.execution_id).await.unwrap();
debug!("Execution history after tick: {execution_log:?}");
let actually_created_at = assert_matches!(
execution_log.events.first().unwrap(),
ExecutionEvent {
event: ExecutionRequest::Created { .. },
created_at: actually_created_at,
backtrace_id: None,
version: Version(0),
}
=> *actually_created_at
);
assert_eq!(config.created_at, actually_created_at);
let locked_at = assert_matches!(
execution_log.events.get(1).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Locked { .. },
created_at: locked_at,
backtrace_id: None,
version: Version(1),
} if config.created_at <= *locked_at
=> *locked_at
);
assert_matches!(execution_log.events.get(2).unwrap(), ExecutionEvent {
event: _,
created_at: executed_at,
backtrace_id: None,
version: Version(2),
} if *executed_at >= locked_at);
execution_log
}
#[rstest]
#[tokio::test]
async fn activity_trap_should_trigger_an_execution_retry(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
set_up();
let sim_clock = SimClock::default();
let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
let retry_exp_backoff = Duration::from_millis(100);
let retry_config = ComponentRetryConfig {
max_retries: Some(1),
retry_exp_backoff,
};
let exec_config = ExecConfig {
batch_size: 1,
lock_expiry: Duration::from_secs(1),
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config,
locking_strategy,
};
let expected_reason = "error reason";
let expected_detail = "error detail";
let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
WorkerError::ActivityTrap {
reason: expected_reason.to_string(),
trap_kind: concepts::TrapKind::Trap,
detail: Some(expected_detail.to_string()),
version: Version::new(2),
http_client_traces: None,
},
)));
debug!(now = %sim_clock.now(), "Creating an execution that should fail");
let db_connection = db_pool.connection_test().await.unwrap();
let execution_log = create_and_tick(
CreateAndTickConfig {
execution_id: ExecutionId::generate(),
created_at: sim_clock.now(),
executed_at: sim_clock.now(),
},
sim_clock.clone_box(),
db_connection.as_ref(),
db_pool.clone(),
exec_config.clone(),
worker,
tick_fn,
)
.await;
assert_eq!(3, execution_log.events.len());
{
let (reason, detail, at, expires_at) = assert_matches!(
&execution_log.events.get(2).unwrap(),
ExecutionEvent {
event: ExecutionRequest::TemporarilyFailed {
reason,
detail,
backoff_expires_at,
http_client_traces: None,
},
created_at: at,
backtrace_id: None,
version: Version(2),
}
=> (reason, detail, *at, *backoff_expires_at)
);
assert_eq!(format!("activity trap: {expected_reason}"), reason.deref());
assert_eq!(Some(expected_detail), detail.as_deref());
assert_eq!(at, sim_clock.now());
assert_eq!(sim_clock.now() + retry_config.retry_exp_backoff, expires_at);
}
let worker = Arc::new(SimpleWorker::with_worker_results_rev(Arc::new(
std::sync::Mutex::new(IndexMap::from([(
Version::new(4),
(
vec![],
WorkerResult::Ok(WorkerResultOk::RunFinished {
retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
version: Version::new(4),
http_client_traces: None,
}),
),
)])),
)));
assert!(
tick_fn(
exec_config.clone(),
sim_clock.clone_box(),
db_pool.clone(),
worker.clone(),
sim_clock.now(),
)
.await
.is_empty()
);
sim_clock.move_time_forward(retry_config.retry_exp_backoff);
tick_fn(
exec_config,
sim_clock.clone_box(),
db_pool.clone(),
worker,
sim_clock.now(),
)
.await;
let execution_log = {
let db_connection = db_pool.connection_test().await.unwrap();
db_connection
.get(&execution_log.execution_id)
.await
.unwrap()
};
debug!(now = %sim_clock.now(), "Execution history after second tick: {execution_log:?}");
assert_matches!(
execution_log.events.get(3).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Locked { .. },
created_at: at,
backtrace_id: None,
version: Version(3),
} if *at == sim_clock.now()
);
assert_matches!(
execution_log.events.get(4).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Finished {
retval: SupportedFunctionReturnValue::Ok(None),
http_client_traces: None
},
created_at: finished_at,
backtrace_id: None,
version: Version(4),
} if *finished_at == sim_clock.now()
);
db_close.close().await;
}
#[rstest]
#[tokio::test]
async fn activity_trap_should_not_be_retried_if_no_retries_are_set(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
set_up();
let created_at = Now.now();
let clock_fn = Box::new(ConstClock(created_at));
let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
let exec_config = ExecConfig {
batch_size: 1,
lock_expiry: Duration::from_secs(1),
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config: ComponentRetryConfig::ZERO,
locking_strategy,
};
let reason = "error reason";
let expected_reason = format!("activity trap: {reason}");
let expected_detail = "error detail";
let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
WorkerError::ActivityTrap {
reason: reason.to_string(),
trap_kind: concepts::TrapKind::Trap,
detail: Some(expected_detail.to_string()),
version: Version::new(2),
http_client_traces: None,
},
)));
let execution_log = create_and_tick(
CreateAndTickConfig {
execution_id: ExecutionId::generate(),
created_at,
executed_at: created_at,
},
clock_fn,
db_pool.connection_test().await.unwrap().as_ref(),
db_pool.clone(),
exec_config.clone(),
worker,
tick_fn,
)
.await;
assert_eq!(3, execution_log.events.len());
let (reason, kind, detail) = assert_matches!(
&execution_log.events.get(2).unwrap(),
ExecutionEvent {
event: ExecutionRequest::Finished{
retval: SupportedFunctionReturnValue::ExecutionError(FinishedExecutionError{reason, kind, detail}),
http_client_traces: None
},
created_at: at,
backtrace_id: None,
version: Version(2),
} if *at == created_at
=> (reason, kind, detail)
);
assert_eq!(Some(expected_reason), *reason);
assert_eq!(Some(expected_detail), detail.as_deref());
assert_eq!(ExecutionFailureKind::Uncategorized, *kind);
db_close.close().await;
}
#[rstest]
#[tokio::test]
async fn child_execution_permanently_failed_should_notify_parent_permanent_failure(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
let worker_error = WorkerError::ActivityTrap {
reason: "error reason".to_string(),
trap_kind: TrapKind::Trap,
detail: Some("detail".to_string()),
version: Version::new(2),
http_client_traces: None,
};
let expected_child_err = FinishedExecutionError {
kind: ExecutionFailureKind::Uncategorized,
reason: Some("activity trap: error reason".to_string()),
detail: Some("detail".to_string()),
};
child_execution_permanently_failed_should_notify_parent(
WorkerResult::Err(worker_error),
expected_child_err,
locking_strategy,
)
.await;
}
#[rstest]
#[tokio::test]
async fn child_execution_permanently_failed_handled_by_watcher_should_notify_parent_timeout(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
let expected_child_err = FinishedExecutionError {
kind: ExecutionFailureKind::TimedOut,
reason: None,
detail: None,
};
child_execution_permanently_failed_should_notify_parent(
WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher),
expected_child_err,
locking_strategy,
)
.await;
}
async fn child_execution_permanently_failed_should_notify_parent(
worker_result: WorkerResult,
expected_child_err: FinishedExecutionError,
locking_strategy: LockingStrategy,
) {
use concepts::storage::JoinSetResponseEventOuter;
const LOCK_EXPIRY: Duration = Duration::from_secs(1);
set_up();
let sim_clock = SimClock::default();
let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
let parent_worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
WorkerResultOk::DbUpdatedByWorkerOrWatcher,
)));
let parent_execution_id = ExecutionId::generate();
db_pool
.connection()
.await
.unwrap()
.create(CreateRequest {
created_at: sim_clock.now(),
execution_id: parent_execution_id.clone(),
ffqn: FFQN_SOME,
params: Params::empty(),
parent: None,
metadata: concepts::ExecutionMetadata::empty(),
scheduled_at: sim_clock.now(),
component_id: ComponentId::dummy_activity(),
deployment_id: DEPLOYMENT_ID_DUMMY,
scheduled_by: None,
})
.await
.unwrap();
let parent_executor_id = ExecutorId::generate();
tick_fn(
ExecConfig {
batch_size: 1,
lock_expiry: LOCK_EXPIRY,
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: parent_executor_id,
retry_config: ComponentRetryConfig::ZERO,
locking_strategy,
},
sim_clock.clone_box(),
db_pool.clone(),
parent_worker,
sim_clock.now(),
)
.await;
let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
let child_execution_id = parent_execution_id.next_level(&join_set_id);
{
let params = Params::empty();
let child = CreateRequest {
created_at: sim_clock.now(),
execution_id: ExecutionId::Derived(child_execution_id.clone()),
ffqn: FFQN_CHILD,
params: params.clone(),
parent: Some((parent_execution_id.clone(), join_set_id.clone())),
metadata: concepts::ExecutionMetadata::empty(),
scheduled_at: sim_clock.now(),
component_id: ComponentId::dummy_activity(),
deployment_id: DEPLOYMENT_ID_DUMMY,
scheduled_by: None,
};
let current_time = sim_clock.now();
let join_set = AppendRequest {
created_at: current_time,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinSetCreate {
join_set_id: join_set_id.clone(),
},
},
};
let child_exec_req = AppendRequest {
created_at: current_time,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinSetRequest {
join_set_id: join_set_id.clone(),
request: JoinSetRequest::ChildExecutionRequest {
child_execution_id: child_execution_id.clone(),
target_ffqn: FFQN_CHILD,
params,
result: Ok(()),
},
},
},
};
let join_next = AppendRequest {
created_at: current_time,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinNext {
join_set_id: join_set_id.clone(),
run_expires_at: sim_clock.now(),
closing: false,
requested_ffqn: Some(FFQN_CHILD),
},
},
};
db_pool
.connection()
.await
.unwrap()
.append_batch_create_new_execution(
current_time,
vec![join_set, child_exec_req, join_next],
parent_execution_id.clone(),
Version::new(2),
vec![child],
vec![],
)
.await
.unwrap();
}
let child_worker =
Arc::new(SimpleWorker::with_single_result(worker_result).with_ffqn(FFQN_CHILD));
tick_fn(
ExecConfig {
batch_size: 1,
lock_expiry: LOCK_EXPIRY,
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config: ComponentRetryConfig::ZERO,
locking_strategy,
},
sim_clock.clone_box(),
db_pool.clone(),
child_worker,
sim_clock.now(),
)
.await;
if matches!(expected_child_err.kind, ExecutionFailureKind::TimedOut) {
sim_clock.move_time_forward(LOCK_EXPIRY);
expired_timers_watcher::tick(
db_pool.connection().await.unwrap().as_ref(),
sim_clock.now(),
)
.await
.unwrap();
}
let child_log = db_pool
.connection_test()
.await
.unwrap()
.get(&ExecutionId::Derived(child_execution_id.clone()))
.await
.unwrap();
assert!(child_log.pending_state.is_finished());
assert_eq!(
Version(2),
child_log.next_version,
"created = 0, locked = 1, with_single_result = 2"
);
assert_eq!(
ExecutionRequest::Finished {
retval: SupportedFunctionReturnValue::ExecutionError(expected_child_err),
http_client_traces: None
},
child_log.last_event().event
);
let parent_log = db_pool
.connection_test()
.await
.unwrap()
.get(&parent_execution_id)
.await
.unwrap();
assert_matches!(
parent_log.pending_state,
PendingState::PendingAt(PendingStatePendingAt {
scheduled_at,
last_lock: Some(LockedBy { executor_id: found_executor_id, run_id: _}),
}) if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
"parent should be back to pending"
);
let (found_join_set_id, found_child_execution_id, child_finished_version, found_result) = assert_matches!(
parent_log.responses.last().map(|resp| &resp.event),
Some(JoinSetResponseEventOuter{
created_at: at,
event: JoinSetResponseEvent{
join_set_id: found_join_set_id,
event: JoinSetResponse::ChildExecutionFinished {
child_execution_id: found_child_execution_id,
finished_version,
result: found_result,
}
}
})
if *at == sim_clock.now()
=> (found_join_set_id, found_child_execution_id, finished_version, found_result)
);
assert_eq!(join_set_id, *found_join_set_id);
assert_eq!(child_execution_id, *found_child_execution_id);
assert_eq!(child_log.next_version, *child_finished_version);
assert_matches!(
found_result,
SupportedFunctionReturnValue::ExecutionError(_)
);
db_close.close().await;
}
#[derive(Clone, Debug)]
struct SleepyWorker {
duration: Duration,
result: SupportedFunctionReturnValue,
exported: [FunctionMetadata; 1],
}
#[async_trait]
impl Worker for SleepyWorker {
async fn run(&self, ctx: WorkerContext) -> WorkerResult {
tokio::time::sleep(self.duration).await;
WorkerResult::Ok(WorkerResultOk::RunFinished {
retval: self.result.clone(),
version: ctx.version,
http_client_traces: None,
})
}
fn exported_functions_noext(&self) -> &[FunctionMetadata] {
&self.exported
}
}
#[rstest]
#[tokio::test]
async fn hanging_lock_should_be_cleaned_and_execution_retried(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
set_up();
let sim_clock = SimClock::default();
let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
let lock_expiry = Duration::from_millis(100);
let timeout_duration = Duration::from_millis(300);
let retry_config = ComponentRetryConfig {
max_retries: Some(1),
retry_exp_backoff: timeout_duration,
};
let exec_config = ExecConfig {
batch_size: 1,
lock_expiry,
tick_sleep: Duration::ZERO,
component_id: ComponentId::dummy_activity(),
task_limiter: None,
executor_id: ExecutorId::generate(),
retry_config,
locking_strategy,
};
let worker = Arc::new(SleepyWorker {
duration: lock_expiry + Duration::from_millis(1), result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
exported: [FunctionMetadata {
ffqn: FFQN_SOME,
parameter_types: ParameterTypes::default(),
return_type: RETURN_TYPE_DUMMY,
extension: None,
submittable: true,
}],
});
let execution_id = ExecutionId::generate();
let db_connection = db_pool.connection_test().await.unwrap();
db_connection
.create(CreateRequest {
created_at: sim_clock.now(),
execution_id: execution_id.clone(),
ffqn: FFQN_SOME,
params: Params::empty(),
parent: None,
metadata: concepts::ExecutionMetadata::empty(),
scheduled_at: sim_clock.now(),
component_id: ComponentId::dummy_activity(),
deployment_id: DEPLOYMENT_ID_DUMMY,
scheduled_by: None,
})
.await
.unwrap();
let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
let executor = ExecTask::new_test(
exec_config.clone(),
worker,
sim_clock.clone_box(),
db_pool.clone(),
ffqns,
);
let db_exec = db_pool.db_exec_conn().await.unwrap();
let mut first_execution_progress = executor
.tick(
db_exec.as_ref(),
sim_clock.now(),
RunId::generate(),
DEPLOYMENT_ID_DUMMY,
)
.await
.unwrap();
assert_eq!(1, first_execution_progress.executions.len());
sim_clock.move_time_forward(lock_expiry);
let now_after_first_lock_expiry = sim_clock.now();
{
debug!(now = %now_after_first_lock_expiry, "Expecting an expired lock");
let cleanup_progress = executor
.tick(
db_pool.db_exec_conn().await.unwrap().as_ref(),
now_after_first_lock_expiry,
RunId::generate(),
DEPLOYMENT_ID_DUMMY,
)
.await
.unwrap();
assert!(cleanup_progress.executions.is_empty());
}
{
let expired_locks = expired_timers_watcher::tick(
db_pool.connection().await.unwrap().as_ref(),
now_after_first_lock_expiry,
)
.await
.unwrap()
.expired_locks;
assert_eq!(1, expired_locks);
}
assert!(
!first_execution_progress
.executions
.pop()
.unwrap()
.1
.is_finished()
);
let execution_log = db_connection.get(&execution_id).await.unwrap();
let expected_first_timeout_expiry = now_after_first_lock_expiry + timeout_duration;
assert_matches!(
&execution_log.events.get(2).unwrap(),
ExecutionEvent {
event: ExecutionRequest::TemporarilyTimedOut { backoff_expires_at, .. },
created_at: at,
backtrace_id: None,
version: Version(2),
} if *at == now_after_first_lock_expiry && *backoff_expires_at == expected_first_timeout_expiry
);
assert_matches!(
execution_log.pending_state,
PendingState::PendingAt(PendingStatePendingAt {
scheduled_at: found_scheduled_by,
last_lock: Some(LockedBy {
executor_id: found_executor_id,
run_id: _,
}),
}) if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
);
sim_clock.move_time_forward(timeout_duration);
let now_after_first_timeout = sim_clock.now();
debug!(now = %now_after_first_timeout, "Second execution should hang again and result in a permanent timeout");
let mut second_execution_progress = executor
.tick(
db_pool.db_exec_conn().await.unwrap().as_ref(),
now_after_first_timeout,
RunId::generate(),
DEPLOYMENT_ID_DUMMY,
)
.await
.unwrap();
assert_eq!(1, second_execution_progress.executions.len());
sim_clock.move_time_forward(lock_expiry);
let now_after_second_lock_expiry = sim_clock.now();
debug!(now = %now_after_second_lock_expiry, "Expecting the second lock to be expired");
{
let cleanup_progress = executor
.tick(
db_pool.db_exec_conn().await.unwrap().as_ref(),
now_after_second_lock_expiry,
RunId::generate(),
DEPLOYMENT_ID_DUMMY,
)
.await
.unwrap();
assert!(cleanup_progress.executions.is_empty());
}
{
let expired_locks = expired_timers_watcher::tick(
db_pool.connection().await.unwrap().as_ref(),
now_after_second_lock_expiry,
)
.await
.unwrap()
.expired_locks;
assert_eq!(1, expired_locks);
}
assert!(
!second_execution_progress
.executions
.pop()
.unwrap()
.1
.is_finished()
);
drop(db_connection);
drop(executor);
db_close.close().await;
}
}