1use crate::worker::{FatalError, Worker, WorkerContext, WorkerError, WorkerResult, WorkerResultOk};
2use assert_matches::assert_matches;
3use chrono::{DateTime, Utc};
4use concepts::prefixed_ulid::{DeploymentId, RunId};
5use concepts::storage::{
6 AppendEventsToExecution, AppendRequest, AppendResponseToExecution, DbErrorGeneric,
7 DbErrorWrite, DbExecutor, DbPool, ExecutionLog, LockedExecution,
8};
9use concepts::time::{ClockFn, Sleep};
10use concepts::{
11 ComponentId, ComponentRetryConfig, ComponentType, FunctionMetadata, StrVariant,
12 SupportedFunctionReturnValue,
13};
14use concepts::{ExecutionFailureKind, JoinSetId};
15use concepts::{ExecutionId, FunctionFqn, prefixed_ulid::ExecutorId};
16use concepts::{
17 FinishedExecutionError,
18 storage::{ExecutionRequest, Version},
19};
20use std::{
21 sync::{
22 Arc,
23 atomic::{AtomicBool, Ordering},
24 },
25 time::Duration,
26};
27use tokio::task::{AbortHandle, JoinHandle};
28use tracing::{Instrument, Level, Span, debug, error, info, info_span, instrument, trace, warn};
29
30#[derive(Debug, Clone)]
31pub struct ExecConfig {
32 pub lock_expiry: Duration,
33 pub tick_sleep: Duration,
34 pub batch_size: u32,
35 pub component_id: ComponentId,
36 pub task_limiter: Option<Arc<tokio::sync::Semaphore>>,
37 pub executor_id: ExecutorId,
38 pub retry_config: ComponentRetryConfig,
39 pub locking_strategy: LockingStrategy,
40}
41
42pub struct ExecTask {
43 worker: Arc<dyn Worker>,
44 pub config: ExecConfig,
45 clock_fn: Box<dyn ClockFn>, db_pool: Arc<dyn DbPool>,
47 locking_strategy_holder: LockingStrategyHolder,
48 worker_count_tx: tokio::sync::watch::Sender<usize>,
49 executor_close_watcher: tokio::sync::watch::Receiver<bool>,
50}
51
52#[derive(derive_more::Debug, Default)]
53pub struct ExecutionProgress {
54 #[debug(skip)]
55 #[allow(dead_code)]
56 executions: Vec<(ExecutionId, JoinHandle<()>)>,
57}
58
59impl ExecutionProgress {
60 #[cfg(feature = "test")]
61 pub async fn wait_for_tasks(self) -> Vec<ExecutionId> {
62 let mut vec = Vec::new();
63 for (exe, join_handle) in self.executions {
64 vec.push(exe);
65 join_handle.await.unwrap();
66 }
67 vec
68 }
69}
70
71#[derive(Clone, Copy, Debug, PartialEq, Eq)]
72pub enum WorkerType {
73 Activity,
74 Workflow,
75}
76
77#[derive(derive_more::Debug)]
78pub struct ExecutorTaskHandle {
79 #[debug(skip)]
80 is_closing: Arc<AtomicBool>,
81 #[debug(skip)]
82 abort_handle: AbortHandle,
83 component_id: ComponentId,
84 executor_id: ExecutorId,
85 deployment_id: DeploymentId,
86 executor_closing_signal_sender: tokio::sync::watch::Sender<bool>,
87 worker_count_rx: tokio::sync::watch::Receiver<usize>,
89}
90
91impl ExecutorTaskHandle {
92 #[instrument(name = "executor.close", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id,
93 deployment_id = %self.deployment_id))]
94 pub async fn close(&self) {
95 trace!("Gracefully closing");
96 self.is_closing.store(true, Ordering::Relaxed);
97 while !self.abort_handle.is_finished() {
98 tokio::time::sleep(Duration::from_millis(1)).await;
99 }
100 trace!("Signaling workflow tasks to unlock");
101 let _ = self.executor_closing_signal_sender.send(true);
102 let mut worker_count_rx = self.worker_count_rx.clone();
103 loop {
104 tokio::select! {
105 () = tokio::time::sleep(Duration::from_secs(1)) => {
106 debug!("Waiting for {} workers to shut down", *self.worker_count_rx.borrow());
107 }
108 _ = worker_count_rx.wait_for(|&count| count == 0) => {
109 break;
110 }
111 }
112 }
113 debug!("Gracefully closed");
114 }
115
116 #[must_use]
117 pub fn component_id(&self) -> &ComponentId {
118 &self.component_id
119 }
120}
121
122impl Drop for ExecutorTaskHandle {
123 #[instrument(level = Level::DEBUG, name = "executor.drop", skip_all, fields(executor_id = %self.executor_id, component_id = %self.component_id))]
124 fn drop(&mut self) {
125 if self.abort_handle.is_finished() {
126 return;
127 }
128 warn!("Aborting the executor task");
129 self.abort_handle.abort();
130 }
131}
132
133#[cfg(feature = "test")]
134pub fn extract_exported_ffqns_noext_test(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
135 extract_exported_ffqns_noext(worker)
136}
137
138fn extract_exported_ffqns_noext(worker: &dyn Worker) -> Arc<[FunctionFqn]> {
139 worker
140 .exported_functions_noext()
141 .iter()
142 .map(|FunctionMetadata { ffqn, .. }| ffqn.clone())
143 .collect::<Arc<_>>()
144}
145
146#[derive(Debug, Clone, Copy)]
147pub enum LockingStrategy {
148 ByFfqns,
149 ByComponentDigest,
150}
151impl LockingStrategy {
152 fn holder(&self, ffqns: Arc<[FunctionFqn]>) -> LockingStrategyHolder {
153 match self {
154 LockingStrategy::ByFfqns => LockingStrategyHolder::ByFfqns(ffqns),
155 LockingStrategy::ByComponentDigest => LockingStrategyHolder::ByComponentId,
156 }
157 }
158}
159
160enum LockingStrategyHolder {
161 ByFfqns(Arc<[FunctionFqn]>),
162 ByComponentId,
163}
164
165impl ExecTask {
166 #[cfg(feature = "test")]
167 pub fn new_test(
168 config: ExecConfig,
169 worker: Arc<dyn Worker>,
170 clock_fn: Box<dyn ClockFn>,
171 db_pool: Arc<dyn DbPool>,
172 ffqns: Arc<[FunctionFqn]>,
173 ) -> Self {
174 let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
175 ExecTask {
176 worker,
177 locking_strategy_holder: config.locking_strategy.holder(ffqns),
178 config,
179 clock_fn,
180 db_pool,
181 worker_count_tx,
182 executor_close_watcher: tokio::sync::watch::channel(false).1,
183 }
184 }
185
186 #[cfg(feature = "test")]
187 pub fn new_all_ffqns_test(
188 worker: Arc<dyn Worker>,
189 config: ExecConfig,
190 clock_fn: Box<dyn ClockFn>,
191 db_pool: Arc<dyn DbPool>,
192 ) -> Self {
193 let ffqns = extract_exported_ffqns_noext(worker.as_ref());
194 let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
195 Self {
196 worker,
197 locking_strategy_holder: config.locking_strategy.holder(ffqns),
198 config,
199 clock_fn,
200 db_pool,
201 worker_count_tx,
202 executor_close_watcher: tokio::sync::watch::channel(false).1,
203 }
204 }
205
206 #[cfg(feature = "test")]
207 pub fn new_all_ffqns_test_with_close_watcher(
208 worker: Arc<dyn Worker>,
209 config: ExecConfig,
210 clock_fn: Box<dyn ClockFn>,
211 db_pool: Arc<dyn DbPool>,
212 executor_close_watcher: tokio::sync::watch::Receiver<bool>,
213 ) -> Self {
214 let ffqns = extract_exported_ffqns_noext(worker.as_ref());
215 let (worker_count_tx, _) = tokio::sync::watch::channel(0usize);
216 Self {
217 worker,
218 locking_strategy_holder: config.locking_strategy.holder(ffqns),
219 config,
220 clock_fn,
221 db_pool,
222 worker_count_tx,
223 executor_close_watcher,
224 }
225 }
226
227 pub fn spawn_new(
228 deployment_id: DeploymentId,
229 worker: Arc<dyn Worker>,
230 config: ExecConfig,
231 clock_fn: Box<dyn ClockFn>,
232 db_pool: Arc<dyn DbPool>,
233 sleep: impl Sleep + Clone + 'static,
234 ) -> ExecutorTaskHandle {
235 let is_closing = Arc::new(AtomicBool::default());
236 let is_closing_inner = is_closing.clone();
237 let ffqns = extract_exported_ffqns_noext(worker.as_ref());
238 let component_id = config.component_id.clone();
239 let executor_id = config.executor_id;
240 let (worker_count_tx, worker_count_rx) = tokio::sync::watch::channel(0);
241 let (executor_closing_signal_sender, executor_close_watcher) =
242 tokio::sync::watch::channel(false);
243 let abort_handle = tokio::spawn(async move {
244 debug!(executor_id = %config.executor_id, component_id = %config.component_id, "Spawned executor");
245 let lock_strategy_holder = config.locking_strategy.holder(ffqns);
246 let task = ExecTask {
247 worker,
248 config,
249 db_pool,
250 locking_strategy_holder: lock_strategy_holder,
251 clock_fn: clock_fn.clone_box(),
252 worker_count_tx,
253 executor_close_watcher,
254 };
255 let mut old_err = None;
256 while !is_closing_inner.load(Ordering::Relaxed) {
257 let res = task.db_pool.db_exec_conn().await;
258 let res = log_err_if_new(res, &mut old_err);
259 if let Ok(db_exec) = res {
260 let _ = task.tick(db_exec.as_ref(), clock_fn.now(), RunId::generate(), deployment_id).await;
261 db_exec
262 .wait_for_pending_by_component_digest(clock_fn.now(), &task.config.component_id.component_digest, {
263 let sleep = sleep.clone();
264 Box::pin(async move { sleep.sleep(task.config.tick_sleep).await })})
265 .await;
266 } else {
267 sleep.sleep(task.config.tick_sleep).await;
268 }
269 }
270 })
271 .abort_handle();
272 ExecutorTaskHandle {
273 is_closing,
274 abort_handle,
275 component_id,
276 executor_id,
277 deployment_id,
278 executor_closing_signal_sender,
279 worker_count_rx,
280 }
281 }
282
283 fn acquire_task_permits(&self) -> Vec<Option<tokio::sync::OwnedSemaphorePermit>> {
284 if let Some(task_limiter) = &self.config.task_limiter {
285 let mut locks = Vec::new();
286 for _ in 0..self.config.batch_size {
287 if let Ok(permit) = task_limiter.clone().try_acquire_owned() {
288 locks.push(Some(permit));
289 } else {
290 break;
291 }
292 }
293 locks
294 } else {
295 let mut vec = Vec::with_capacity(self.config.batch_size as usize);
296 for _ in 0..self.config.batch_size {
297 vec.push(None);
298 }
299 vec
300 }
301 }
302
303 #[cfg(feature = "test")]
304 pub async fn tick_test(&self, executed_at: DateTime<Utc>, run_id: RunId) -> ExecutionProgress {
305 use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
306
307 let db_exec = self.db_pool.db_exec_conn().await.unwrap();
308 self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
309 .await
310 .unwrap()
311 }
312
313 #[cfg(feature = "test")]
314 pub async fn tick_test_await(
315 &self,
316 executed_at: DateTime<Utc>,
317 run_id: RunId,
318 ) -> Vec<ExecutionId> {
319 use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
320
321 let db_exec = self.db_pool.db_exec_conn().await.unwrap();
322 self.tick(db_exec.as_ref(), executed_at, run_id, DEPLOYMENT_ID_DUMMY)
323 .await
324 .unwrap()
325 .wait_for_tasks()
326 .await
327 }
328
329 #[instrument(level = Level::TRACE, name = "executor.tick" skip_all, fields(executor_id = %self.config.executor_id, component_id = %self.config.component_id))]
330 async fn tick(
331 &self,
332 db_exec: &dyn DbExecutor,
333 executed_at: DateTime<Utc>,
334 run_id: RunId,
335 deployment_id: DeploymentId,
336 ) -> Result<ExecutionProgress, DbErrorWrite> {
337 let locked_executions = {
338 let mut permits = self.acquire_task_permits();
339 if permits.is_empty() {
340 return Ok(ExecutionProgress::default());
341 }
342 let lock_expires_at = executed_at + self.config.lock_expiry;
343 let batch_size = u32::try_from(permits.len()).expect("ExecConfig.batch_size is u32");
344 let locked_executions = match &self.locking_strategy_holder {
345 LockingStrategyHolder::ByFfqns(ffqns) => {
346 db_exec
347 .lock_pending_by_ffqns(
348 batch_size,
349 executed_at, ffqns.clone(),
351 executed_at, self.config.component_id.clone(),
353 deployment_id,
354 self.config.executor_id,
355 lock_expires_at,
356 run_id,
357 self.config.retry_config,
358 )
359 .await?
360 }
361 LockingStrategyHolder::ByComponentId => {
362 db_exec
363 .lock_pending_by_component_digest(
364 batch_size,
365 executed_at, &self.config.component_id,
367 deployment_id,
368 executed_at, self.config.executor_id,
370 lock_expires_at,
371 run_id,
372 self.config.retry_config,
373 )
374 .await?
375 }
376 };
377 while permits.len() > locked_executions.len() {
379 permits.pop();
380 }
381 assert_eq!(permits.len(), locked_executions.len());
382 locked_executions.into_iter().zip(permits)
383 };
384
385 let mut executions = Vec::with_capacity(locked_executions.len());
386 for (locked_execution, permit) in locked_executions {
387 let execution_id = locked_execution.execution_id.clone();
388 let join_handle = {
389 let worker = self.worker.clone();
390 let db_pool = self.db_pool.clone();
391 let clock_fn = self.clock_fn.clone_box();
392 let worker_span = info_span!(parent: None, "worker",
393 "otel.name" = format!("worker {}", locked_execution.ffqn),
394 %execution_id, %run_id,
395 ffqn = %locked_execution.ffqn,
396 executor_id = %self.config.executor_id,
397 component_id = %self.config.component_id,
398 %deployment_id,
399 );
400 locked_execution.metadata.enrich(&worker_span);
401 let component_type = self.config.component_id.component_type;
402 let worker_count_tx = self.worker_count_tx.clone();
403 worker_count_tx.send_modify(|n| *n += 1);
404 let executor_close_watcher = self.executor_close_watcher.clone();
405 tokio::spawn({
406 let worker_span2 = worker_span.clone();
407 let retry_config = self.config.retry_config;
408 async move {
409 let _permit = permit;
410 let res = Self::run_worker(
411 component_type,
412 worker,
413 db_pool.as_ref(),
414 clock_fn,
415 locked_execution,
416 retry_config,
417 worker_span2,
418 executor_close_watcher
419 )
420 .await;
421 if let Err(db_error) = res {
422 error!("Got db error `{db_error:?}`, expecting watcher to mark execution as timed out");
423 }
424 worker_count_tx.send_modify(|n| *n -= 1);
425 }
426 .instrument(worker_span)
427 })
428 };
429 executions.push((execution_id, join_handle));
430 }
431 Ok(ExecutionProgress { executions })
432 }
433
434 #[expect(clippy::too_many_arguments)]
435 async fn run_worker(
436 component_type: ComponentType,
437 worker: Arc<dyn Worker>,
438 db_pool: &dyn DbPool,
439 clock_fn: Box<dyn ClockFn>,
440 locked_execution: LockedExecution,
441 retry_config: ComponentRetryConfig,
442 worker_span: Span,
443 executor_close_watcher: tokio::sync::watch::Receiver<bool>,
444 ) -> Result<(), DbErrorWrite> {
445 debug!("Worker::run starting");
446 trace!(
447 version = %locked_execution.next_version,
448 params = ?locked_execution.params,
449 event_history = ?locked_execution.event_history,
450 "Worker::run starting"
451 );
452 let can_be_retried = ExecutionLog::can_be_retried_after(
453 locked_execution.intermittent_event_count + 1,
454 retry_config.max_retries,
455 retry_config.retry_exp_backoff,
456 );
457 let unlock_expiry_on_limit_reached =
458 ExecutionLog::compute_retry_duration_when_retrying_forever(
459 locked_execution.intermittent_event_count + 1,
460 retry_config.retry_exp_backoff,
461 );
462 let ctx = WorkerContext {
463 execution_id: locked_execution.execution_id.clone(),
464 metadata: locked_execution.metadata,
465 ffqn: locked_execution.ffqn,
466 params: locked_execution.params,
467 event_history: locked_execution.event_history,
468 responses: locked_execution.responses,
469 version: locked_execution.next_version,
470 can_be_retried: can_be_retried.is_some(),
471 locked_event: locked_execution.locked_event,
472 worker_span,
473 executor_close_watcher,
474 };
475 let worker_result = worker.run(ctx).await;
476 debug!("Worker::run finished {worker_result:?}");
477 let result_obtained_at = clock_fn.now();
478 match Self::worker_result_to_execution_event(
479 component_type,
480 locked_execution.execution_id,
481 worker_result,
482 result_obtained_at,
483 locked_execution.parent,
484 can_be_retried,
485 unlock_expiry_on_limit_reached,
486 )? {
487 Some(append) => {
488 trace!("Appending {append:?}");
489 let db_exec = db_pool.db_exec_conn().await?;
490 match append {
491 AppendOrCancel::Cancel {
492 execution_id,
493 cancelled_at,
494 } => db_exec
495 .cancel_activity_with_retries(&execution_id, cancelled_at)
496 .await
497 .map(|_| ()),
498 AppendOrCancel::Other(append) => append.append(db_exec.as_ref()).await,
499 }
500 }
501 None => Ok(()),
502 }
503 }
504
505 fn worker_result_to_execution_event(
507 component_type: ComponentType,
508 execution_id: ExecutionId,
509 worker_result: WorkerResult,
510 result_obtained_at: DateTime<Utc>,
511 parent: Option<(ExecutionId, JoinSetId)>,
512 can_be_retried: Option<Duration>,
513 unlock_expiry_on_limit_reached: Duration,
514 ) -> Result<Option<AppendOrCancel>, DbErrorWrite> {
515 Ok(match worker_result {
516 WorkerResult::Ok(WorkerResultOk::RunFinished {
517 retval: ref retval @ SupportedFunctionReturnValue::Err(ref result_err),
518 version,
519 http_client_traces,
520 }) if component_type == ComponentType::Activity
521 && can_be_retried.is_some()
522 && !retval.is_permanent_variant() =>
523 {
524 let detail = serde_json::to_string(result_err)
526 .expect("SupportedFunctionReturnValue should be serializable to JSON");
527 let duration = can_be_retried.expect(
528 "ActivityReturnedError must not be returned when retries are exhausted",
529 );
530 let expires_at = result_obtained_at + duration;
531 debug!("Retrying ActivityReturnedError after {duration:?} at {expires_at}");
532 let primary_event = ExecutionRequest::TemporarilyFailed {
533 backoff_expires_at: expires_at,
534 reason: StrVariant::Static("activity returned error"),
535 detail: Some(detail),
536 http_client_traces,
537 };
538 Some(AppendOrCancel::Other(Append {
539 created_at: result_obtained_at,
540 primary_event: AppendRequest {
541 created_at: result_obtained_at,
542 event: primary_event,
543 },
544 execution_id,
545 version,
546 child_finished: None,
547 }))
548 }
549
550 WorkerResult::Ok(WorkerResultOk::RunFinished {
551 retval: result,
552 version,
553 http_client_traces,
554 }) => {
555 info!("Execution finished: {result}");
556 let child_finished =
557 parent.map(
558 |(parent_execution_id, parent_join_set)| ChildFinishedResponse {
559 parent_execution_id,
560 parent_join_set,
561 result: result.clone(),
562 },
563 );
564 let primary_event = AppendRequest {
565 created_at: result_obtained_at,
566 event: ExecutionRequest::Finished {
567 retval: result,
568 http_client_traces,
569 },
570 };
571
572 Some(AppendOrCancel::Other(Append {
573 created_at: result_obtained_at,
574 primary_event,
575 execution_id,
576 version,
577 child_finished,
578 }))
579 }
580
581 WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher) => None,
582
583 WorkerResult::Err(err) => {
584 let reason_generic = err.to_string(); let (primary_event, child_finished, version) = match err {
587 WorkerError::ExecutorClosing(version) => {
588 let primary_event = ExecutionRequest::Unlocked {
589 backoff_expires_at: result_obtained_at, reason: "executor closing".into(),
591 };
592 (primary_event, None, version)
593 }
594 WorkerError::TemporaryTimeout {
595 http_client_traces,
596 version,
597 } => {
598 if let Some(duration) = can_be_retried {
599 let backoff_expires_at = result_obtained_at + duration;
600 info!(
601 "Temporary timeout, retrying after {duration:?} at {backoff_expires_at}"
602 );
603 (
604 ExecutionRequest::TemporarilyTimedOut {
605 backoff_expires_at,
606 http_client_traces,
607 },
608 None,
609 version,
610 )
611 } else {
612 info!("Execution timed out");
613 let result = SupportedFunctionReturnValue::ExecutionError(
614 FinishedExecutionError {
615 kind: ExecutionFailureKind::TimedOut,
616 reason: None,
617 detail: None,
618 },
619 );
620 let child_finished =
621 parent.map(|(parent_execution_id, parent_join_set)| {
622 ChildFinishedResponse {
623 parent_execution_id,
624 parent_join_set,
625 result: result.clone(),
626 }
627 });
628 (
629 ExecutionRequest::Finished {
630 retval: result,
631 http_client_traces,
632 },
633 child_finished,
634 version,
635 )
636 }
637 }
638 WorkerError::DbError(db_error) => {
639 return Err(db_error);
640 }
641 WorkerError::ActivityTrap {
642 reason: _, trap_kind,
644 detail,
645 version,
646 http_client_traces,
647 } => {
648 if let Some(duration) = can_be_retried {
649 let expires_at = result_obtained_at + duration;
650 debug!(
651 "Retrying activity with `{trap_kind}` execution after {duration:?} at {expires_at}"
652 );
653 (
654 ExecutionRequest::TemporarilyFailed {
655 reason: StrVariant::from(reason_generic),
656 backoff_expires_at: expires_at,
657 detail,
658 http_client_traces,
659 },
660 None,
661 version,
662 )
663 } else {
664 info!(
665 "Activity with `{trap_kind}` marked as permanent failure - {reason_generic}"
666 );
667 let result = SupportedFunctionReturnValue::ExecutionError(
668 FinishedExecutionError {
669 reason: Some(reason_generic),
670 kind: ExecutionFailureKind::Uncategorized,
671 detail,
672 },
673 );
674 let child_finished =
675 parent.map(|(parent_execution_id, parent_join_set)| {
676 ChildFinishedResponse {
677 parent_execution_id,
678 parent_join_set,
679 result: result.clone(),
680 }
681 });
682 (
683 ExecutionRequest::Finished {
684 retval: result,
685 http_client_traces,
686 },
687 child_finished,
688 version,
689 )
690 }
691 }
692 WorkerError::ActivityPreopenedDirError {
693 reason,
694 detail,
695 version,
696 } => {
697 let http_client_traces = None;
698 if let Some(duration) = can_be_retried {
699 let expires_at = result_obtained_at + duration;
700 debug!(
701 "Retrying activity with ActivityPreopenedDirError `{reason}` execution after {duration:?} at {expires_at}"
702 );
703 (
704 ExecutionRequest::TemporarilyFailed {
705 reason: StrVariant::from(reason_generic),
706 backoff_expires_at: expires_at,
707 detail: Some(detail),
708 http_client_traces,
709 },
710 None,
711 version,
712 )
713 } else {
714 info!(
715 "Activity with ActivityPreopenedDirError `{reason}` marked as permanent failure - {reason_generic}"
716 );
717 let result = SupportedFunctionReturnValue::ExecutionError(
718 FinishedExecutionError {
719 reason: Some(reason_generic),
720 kind: ExecutionFailureKind::Uncategorized,
721 detail: Some(detail),
722 },
723 );
724 let child_finished =
725 parent.map(|(parent_execution_id, parent_join_set)| {
726 ChildFinishedResponse {
727 parent_execution_id,
728 parent_join_set,
729 result: result.clone(),
730 }
731 });
732 (
733 ExecutionRequest::Finished {
734 retval: result,
735 http_client_traces,
736 },
737 child_finished,
738 version,
739 )
740 }
741 }
742 WorkerError::LimitReached {
743 reason,
744 version: new_version,
745 } => {
746 let expires_at = result_obtained_at + unlock_expiry_on_limit_reached;
747 warn!(
748 "Limit reached: {reason}, unlocking after {unlock_expiry_on_limit_reached:?} at {expires_at}"
749 );
750 (
751 ExecutionRequest::Unlocked {
752 backoff_expires_at: expires_at,
753 reason: StrVariant::from(reason),
754 },
755 None,
756 new_version,
757 )
758 }
759 WorkerError::FatalError(FatalError::Cancelled, _version) => {
760 return Ok(Some(AppendOrCancel::Cancel {
762 execution_id,
763 cancelled_at: result_obtained_at,
764 }));
765 }
766 WorkerError::FatalError(fatal_error, version) => {
767 warn!("Fatal worker error - {fatal_error:?}");
768 let result = SupportedFunctionReturnValue::ExecutionError(
769 FinishedExecutionError::from(fatal_error),
770 );
771 let child_finished =
772 parent.map(|(parent_execution_id, parent_join_set)| {
773 ChildFinishedResponse {
774 parent_execution_id,
775 parent_join_set,
776 result: result.clone(),
777 }
778 });
779 (
780 ExecutionRequest::Finished {
781 retval: result,
782 http_client_traces: None,
783 },
784 child_finished,
785 version,
786 )
787 }
788 };
789 Some(AppendOrCancel::Other(Append {
790 created_at: result_obtained_at,
791 primary_event: AppendRequest {
792 created_at: result_obtained_at,
793 event: primary_event,
794 },
795 execution_id,
796 version,
797 child_finished,
798 }))
799 }
800 })
801 }
802}
803
804#[derive(Debug, Clone)]
805pub(crate) struct ChildFinishedResponse {
806 pub(crate) parent_execution_id: ExecutionId,
807 pub(crate) parent_join_set: JoinSetId,
808 pub(crate) result: SupportedFunctionReturnValue,
809}
810
811#[derive(Debug, Clone)]
812#[expect(clippy::large_enum_variant)]
813pub(crate) enum AppendOrCancel {
814 Cancel {
815 execution_id: ExecutionId,
816 cancelled_at: DateTime<Utc>,
817 },
818 Other(Append),
819}
820
821#[derive(Debug, Clone)]
822pub(crate) struct Append {
823 pub(crate) created_at: DateTime<Utc>,
824 pub(crate) primary_event: AppendRequest,
825 pub(crate) execution_id: ExecutionId,
826 pub(crate) version: Version,
827 pub(crate) child_finished: Option<ChildFinishedResponse>,
828}
829
830impl Append {
831 pub(crate) async fn append(self, db_exec: &dyn DbExecutor) -> Result<(), DbErrorWrite> {
832 if let Some(child_finished) = self.child_finished {
833 assert_matches!(
834 &self.primary_event,
835 AppendRequest {
836 event: ExecutionRequest::Finished { .. },
837 ..
838 }
839 );
840 let child_execution_id = assert_matches!(self.execution_id.clone(), ExecutionId::Derived(derived) => derived);
841 let events = AppendEventsToExecution {
842 execution_id: self.execution_id,
843 version: self.version.clone(),
844 batch: vec![self.primary_event],
845 };
846 let response = AppendResponseToExecution {
847 parent_execution_id: child_finished.parent_execution_id,
848 created_at: self.created_at,
849 join_set_id: child_finished.parent_join_set,
850 child_execution_id,
851 finished_version: self.version, result: child_finished.result,
853 };
854
855 db_exec
856 .append_batch_respond_to_parent(events, response, self.created_at)
857 .await?;
858 } else {
859 db_exec
860 .append(self.execution_id, self.version, self.primary_event)
861 .await?;
862 }
863 Ok(())
864 }
865}
866
867fn log_err_if_new<T>(
868 res: Result<T, DbErrorGeneric>,
869 old_err: &mut Option<DbErrorGeneric>,
870) -> Result<T, ()> {
871 match (res, &old_err) {
872 (Ok(ok), _) => {
873 *old_err = None;
874 Ok(ok)
875 }
876 (Err(err), Some(old)) if err == *old => Err(()),
877 (Err(err), _) => {
878 warn!("Tick failed: {err:?}");
879 *old_err = Some(err);
880 Err(())
881 }
882 }
883}
884
885#[cfg(any(test, feature = "test"))]
886pub mod simple_worker {
887 use crate::worker::{Worker, WorkerContext, WorkerResult};
888 use async_trait::async_trait;
889 use concepts::{
890 FunctionFqn, FunctionMetadata, ParameterTypes, RETURN_TYPE_DUMMY,
891 storage::{HistoryEvent, Version},
892 };
893 use indexmap::IndexMap;
894 use std::sync::Arc;
895 use tracing::trace;
896
897 pub(crate) const FFQN_SOME: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn");
898 pub type SimpleWorkerResultMap =
899 Arc<std::sync::Mutex<IndexMap<Version, (Vec<HistoryEvent>, WorkerResult)>>>;
900
901 #[derive(Clone, Debug)]
902 pub struct SimpleWorker {
903 pub worker_results_rev: SimpleWorkerResultMap,
904 pub ffqn: FunctionFqn,
905 exported: [FunctionMetadata; 1],
906 }
907
908 impl SimpleWorker {
909 #[must_use]
910 pub fn with_single_result(res: WorkerResult) -> Self {
911 Self::with_worker_results_rev(Arc::new(std::sync::Mutex::new(IndexMap::from([(
912 Version::new(2),
913 (vec![], res),
914 )]))))
915 }
916
917 #[must_use]
918 pub fn with_ffqn(self, ffqn: FunctionFqn) -> Self {
919 Self {
920 worker_results_rev: self.worker_results_rev,
921 exported: [FunctionMetadata {
922 ffqn: ffqn.clone(),
923 parameter_types: ParameterTypes::default(),
924 return_type: RETURN_TYPE_DUMMY,
925 extension: None,
926 submittable: true,
927 }],
928 ffqn,
929 }
930 }
931
932 #[must_use]
933 pub fn with_worker_results_rev(worker_results_rev: SimpleWorkerResultMap) -> Self {
934 Self {
935 worker_results_rev,
936 ffqn: FFQN_SOME,
937 exported: [FunctionMetadata {
938 ffqn: FFQN_SOME,
939 parameter_types: ParameterTypes::default(),
940 return_type: RETURN_TYPE_DUMMY,
941 extension: None,
942 submittable: true,
943 }],
944 }
945 }
946 }
947
948 #[async_trait]
949 impl Worker for SimpleWorker {
950 async fn run(&self, ctx: WorkerContext) -> WorkerResult {
951 let (expected_version, (expected_eh, worker_result)) =
952 self.worker_results_rev.lock().unwrap().pop().unwrap();
953 trace!(%expected_version, version = %ctx.version, ?expected_eh, eh = ?ctx.event_history, "Running SimpleWorker");
954 assert_eq!(expected_version, ctx.version);
955 assert_eq!(
956 expected_eh,
957 ctx.event_history
958 .iter()
959 .map(|(event, _version)| event.clone())
960 .collect::<Vec<_>>()
961 );
962 worker_result
963 }
964
965 fn exported_functions_noext(&self) -> &[FunctionMetadata] {
966 &self.exported
967 }
968 }
969}
970
971#[cfg(test)]
972mod tests {
973 use self::simple_worker::SimpleWorker;
974 use super::*;
975 use crate::{expired_timers_watcher, worker::WorkerResult};
976 use assert_matches::assert_matches;
977 use async_trait::async_trait;
978 use concepts::prefixed_ulid::DEPLOYMENT_ID_DUMMY;
979 use concepts::storage::{
980 CreateRequest, DbConnectionTest, JoinSetRequest, JoinSetResponse, JoinSetResponseEvent,
981 };
982 use concepts::storage::{DbPoolCloseable, LockedBy};
983 use concepts::storage::{
984 ExecutionEvent, ExecutionRequest, HistoryEvent, PendingState, PendingStatePendingAt,
985 };
986 use concepts::time::{ConstClock, Now};
987 use concepts::{
988 FunctionMetadata, JoinSetKind, ParameterTypes, Params, RETURN_TYPE_DUMMY,
989 SUPPORTED_RETURN_VALUE_OK_EMPTY, StrVariant, SupportedFunctionReturnValue, TrapKind,
990 };
991 use db_tests::Database;
992 use indexmap::IndexMap;
993 use rstest::rstest;
994 use simple_worker::FFQN_SOME;
995 use std::{fmt::Debug, future::Future, ops::Deref, sync::Arc};
996 use test_db_macro::expand_enum_database;
997 use test_utils::set_up;
998 use test_utils::sim_clock::SimClock;
999
1000 pub(crate) const FFQN_CHILD: FunctionFqn = FunctionFqn::new_static("ns:pkg/ifc", "fn-child");
1001
1002 async fn tick_fn<W: Worker + Debug>(
1003 config: ExecConfig,
1004 clock_fn: Box<dyn ClockFn>,
1005 db_pool: Arc<dyn DbPool>,
1006 worker: Arc<W>,
1007 executed_at: DateTime<Utc>,
1008 ) -> Vec<ExecutionId> {
1009 trace!("Ticking with {worker:?}");
1010 let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
1011 let executor = ExecTask::new_test(config, worker, clock_fn, db_pool, ffqns);
1012 executor
1013 .tick_test_await(executed_at, RunId::generate())
1014 .await
1015 }
1016
1017 #[expand_enum_database]
1018 #[rstest]
1019 #[tokio::test]
1020 async fn execute_simple_lifecycle_tick_based(
1021 database: Database,
1022 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1023 locking_strategy: LockingStrategy,
1024 ) {
1025 set_up();
1026 let created_at = Now.now();
1027 let (_guard, db_pool, db_close) = database.set_up().await;
1028 let db_connection = db_pool.connection_test().await.unwrap();
1029 execute_simple_lifecycle_tick_based_inner(
1030 db_connection.as_ref(),
1031 db_pool.clone(),
1032 Box::new(ConstClock(created_at)),
1033 locking_strategy,
1034 )
1035 .await;
1036 drop(db_connection);
1037 db_close.close().await;
1038 }
1039
1040 async fn execute_simple_lifecycle_tick_based_inner(
1041 db_connection: &dyn DbConnectionTest,
1042 db_pool: Arc<dyn DbPool>,
1043 clock_fn: Box<dyn ClockFn>,
1044 locking_strategy: LockingStrategy,
1045 ) {
1046 let created_at = clock_fn.now();
1047 let exec_config = ExecConfig {
1048 batch_size: 1,
1049 lock_expiry: Duration::from_secs(1),
1050 tick_sleep: Duration::from_millis(100),
1051 component_id: ComponentId::dummy_activity(),
1052 task_limiter: None,
1053 executor_id: ExecutorId::generate(),
1054 retry_config: ComponentRetryConfig::ZERO,
1055 locking_strategy,
1056 };
1057
1058 let execution_log = create_and_tick(
1059 CreateAndTickConfig {
1060 execution_id: ExecutionId::generate(),
1061 created_at,
1062 executed_at: created_at,
1063 },
1064 clock_fn,
1065 db_connection,
1066 db_pool,
1067 exec_config,
1068 Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1069 WorkerResultOk::RunFinished {
1070 retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1071 version: Version::new(2),
1072 http_client_traces: None,
1073 },
1074 ))),
1075 tick_fn,
1076 )
1077 .await;
1078 assert_matches!(
1079 execution_log.events.get(2).unwrap(),
1080 ExecutionEvent {
1081 event: ExecutionRequest::Finished {
1082 retval: SupportedFunctionReturnValue::Ok(None),
1083 http_client_traces: None
1084 },
1085 created_at: _,
1086 backtrace_id: None,
1087 version: Version(2),
1088 }
1089 );
1090 }
1091
1092 #[rstest]
1093 #[tokio::test]
1094 async fn execute_simple_lifecycle_task_based_mem(
1095 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1096 locking_strategy: LockingStrategy,
1097 ) {
1098 set_up();
1099 let created_at = Now.now();
1100 let clock_fn = Box::new(ConstClock(created_at));
1101 let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1102 let exec_config = ExecConfig {
1103 batch_size: 1,
1104 lock_expiry: Duration::from_secs(1),
1105 tick_sleep: Duration::ZERO,
1106 component_id: ComponentId::dummy_activity(),
1107 task_limiter: None,
1108 executor_id: ExecutorId::generate(),
1109 retry_config: ComponentRetryConfig::ZERO,
1110 locking_strategy,
1111 };
1112
1113 let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1114 WorkerResultOk::RunFinished {
1115 retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1116 version: Version::new(2),
1117 http_client_traces: None,
1118 },
1119 )));
1120 let db_connection = db_pool.connection_test().await.unwrap();
1121
1122 let execution_log = create_and_tick(
1123 CreateAndTickConfig {
1124 execution_id: ExecutionId::generate(),
1125 created_at,
1126 executed_at: created_at,
1127 },
1128 clock_fn,
1129 db_connection.as_ref(),
1130 db_pool,
1131 exec_config,
1132 worker,
1133 tick_fn,
1134 )
1135 .await;
1136 assert_matches!(
1137 execution_log.events.get(2).unwrap(),
1138 ExecutionEvent {
1139 event: ExecutionRequest::Finished {
1140 retval: SupportedFunctionReturnValue::Ok(None),
1141 http_client_traces: None
1142 },
1143 created_at: _,
1144 backtrace_id: None,
1145 version: Version(2),
1146 }
1147 );
1148 db_close.close().await;
1149 }
1150
1151 struct CreateAndTickConfig {
1152 execution_id: ExecutionId,
1153 created_at: DateTime<Utc>,
1154 executed_at: DateTime<Utc>,
1155 }
1156
1157 async fn create_and_tick<
1158 W: Worker,
1159 T: FnMut(ExecConfig, Box<dyn ClockFn>, Arc<dyn DbPool>, Arc<W>, DateTime<Utc>) -> F,
1160 F: Future<Output = Vec<ExecutionId>>,
1161 >(
1162 config: CreateAndTickConfig,
1163 clock_fn: Box<dyn ClockFn>,
1164 db_connection: &dyn DbConnectionTest,
1165 db_pool: Arc<dyn DbPool>,
1166 exec_config: ExecConfig,
1167 worker: Arc<W>,
1168 mut tick: T,
1169 ) -> ExecutionLog {
1170 db_connection
1172 .create(CreateRequest {
1173 created_at: config.created_at,
1174 execution_id: config.execution_id.clone(),
1175 ffqn: FFQN_SOME,
1176 params: Params::empty(),
1177 parent: None,
1178 metadata: concepts::ExecutionMetadata::empty(),
1179 scheduled_at: config.created_at,
1180 component_id: ComponentId::dummy_activity(),
1181 deployment_id: DEPLOYMENT_ID_DUMMY,
1182 scheduled_by: None,
1183 })
1184 .await
1185 .unwrap();
1186 tick(exec_config, clock_fn, db_pool, worker, config.executed_at).await;
1188 let execution_log = db_connection.get(&config.execution_id).await.unwrap();
1189 debug!("Execution history after tick: {execution_log:?}");
1190 let actually_created_at = assert_matches!(
1192 execution_log.events.first().unwrap(),
1193 ExecutionEvent {
1194 event: ExecutionRequest::Created { .. },
1195 created_at: actually_created_at,
1196 backtrace_id: None,
1197 version: Version(0),
1198 }
1199 => *actually_created_at
1200 );
1201 assert_eq!(config.created_at, actually_created_at);
1202 let locked_at = assert_matches!(
1203 execution_log.events.get(1).unwrap(),
1204 ExecutionEvent {
1205 event: ExecutionRequest::Locked { .. },
1206 created_at: locked_at,
1207 backtrace_id: None,
1208 version: Version(1),
1209 } if config.created_at <= *locked_at
1210 => *locked_at
1211 );
1212 assert_matches!(execution_log.events.get(2).unwrap(), ExecutionEvent {
1213 event: _,
1214 created_at: executed_at,
1215 backtrace_id: None,
1216 version: Version(2),
1217 } if *executed_at >= locked_at);
1218 execution_log
1219 }
1220
1221 #[rstest]
1222 #[tokio::test]
1223 async fn activity_trap_should_trigger_an_execution_retry(
1224 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1225 locking_strategy: LockingStrategy,
1226 ) {
1227 set_up();
1228 let sim_clock = SimClock::default();
1229 let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1230 let retry_exp_backoff = Duration::from_millis(100);
1231 let retry_config = ComponentRetryConfig {
1232 max_retries: Some(1),
1233 retry_exp_backoff,
1234 };
1235 let exec_config = ExecConfig {
1236 batch_size: 1,
1237 lock_expiry: Duration::from_secs(1),
1238 tick_sleep: Duration::ZERO,
1239 component_id: ComponentId::dummy_activity(),
1240 task_limiter: None,
1241 executor_id: ExecutorId::generate(),
1242 retry_config,
1243 locking_strategy,
1244 };
1245 let expected_reason = "error reason";
1246 let expected_detail = "error detail";
1247 let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
1248 WorkerError::ActivityTrap {
1249 reason: expected_reason.to_string(),
1250 trap_kind: concepts::TrapKind::Trap,
1251 detail: Some(expected_detail.to_string()),
1252 version: Version::new(2),
1253 http_client_traces: None,
1254 },
1255 )));
1256 debug!(now = %sim_clock.now(), "Creating an execution that should fail");
1257 let db_connection = db_pool.connection_test().await.unwrap();
1258 let execution_log = create_and_tick(
1259 CreateAndTickConfig {
1260 execution_id: ExecutionId::generate(),
1261 created_at: sim_clock.now(),
1262 executed_at: sim_clock.now(),
1263 },
1264 sim_clock.clone_box(),
1265 db_connection.as_ref(),
1266 db_pool.clone(),
1267 exec_config.clone(),
1268 worker,
1269 tick_fn,
1270 )
1271 .await;
1272 assert_eq!(3, execution_log.events.len());
1273 {
1274 let (reason, detail, at, expires_at) = assert_matches!(
1275 &execution_log.events.get(2).unwrap(),
1276 ExecutionEvent {
1277 event: ExecutionRequest::TemporarilyFailed {
1278 reason,
1279 detail,
1280 backoff_expires_at,
1281 http_client_traces: None,
1282 },
1283 created_at: at,
1284 backtrace_id: None,
1285 version: Version(2),
1286 }
1287 => (reason, detail, *at, *backoff_expires_at)
1288 );
1289 assert_eq!(format!("activity trap: {expected_reason}"), reason.deref());
1290 assert_eq!(Some(expected_detail), detail.as_deref());
1291 assert_eq!(at, sim_clock.now());
1292 assert_eq!(sim_clock.now() + retry_config.retry_exp_backoff, expires_at);
1293 }
1294 let worker = Arc::new(SimpleWorker::with_worker_results_rev(Arc::new(
1295 std::sync::Mutex::new(IndexMap::from([(
1296 Version::new(4),
1297 (
1298 vec![],
1299 WorkerResult::Ok(WorkerResultOk::RunFinished {
1300 retval: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1301 version: Version::new(4),
1302 http_client_traces: None,
1303 }),
1304 ),
1305 )])),
1306 )));
1307 assert!(
1309 tick_fn(
1310 exec_config.clone(),
1311 sim_clock.clone_box(),
1312 db_pool.clone(),
1313 worker.clone(),
1314 sim_clock.now(),
1315 )
1316 .await
1317 .is_empty()
1318 );
1319 sim_clock.move_time_forward(retry_config.retry_exp_backoff);
1321 tick_fn(
1322 exec_config,
1323 sim_clock.clone_box(),
1324 db_pool.clone(),
1325 worker,
1326 sim_clock.now(),
1327 )
1328 .await;
1329 let execution_log = {
1330 let db_connection = db_pool.connection_test().await.unwrap();
1331 db_connection
1332 .get(&execution_log.execution_id)
1333 .await
1334 .unwrap()
1335 };
1336 debug!(now = %sim_clock.now(), "Execution history after second tick: {execution_log:?}");
1337 assert_matches!(
1338 execution_log.events.get(3).unwrap(),
1339 ExecutionEvent {
1340 event: ExecutionRequest::Locked { .. },
1341 created_at: at,
1342 backtrace_id: None,
1343 version: Version(3),
1344 } if *at == sim_clock.now()
1345 );
1346 assert_matches!(
1347 execution_log.events.get(4).unwrap(),
1348 ExecutionEvent {
1349 event: ExecutionRequest::Finished {
1350 retval: SupportedFunctionReturnValue::Ok(None),
1351 http_client_traces: None
1352 },
1353 created_at: finished_at,
1354 backtrace_id: None,
1355 version: Version(4),
1356 } if *finished_at == sim_clock.now()
1357 );
1358 db_close.close().await;
1359 }
1360
1361 #[rstest]
1362 #[tokio::test]
1363 async fn activity_trap_should_not_be_retried_if_no_retries_are_set(
1364 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1365 locking_strategy: LockingStrategy,
1366 ) {
1367 set_up();
1368 let created_at = Now.now();
1369 let clock_fn = Box::new(ConstClock(created_at));
1370 let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1371 let exec_config = ExecConfig {
1372 batch_size: 1,
1373 lock_expiry: Duration::from_secs(1),
1374 tick_sleep: Duration::ZERO,
1375 component_id: ComponentId::dummy_activity(),
1376 task_limiter: None,
1377 executor_id: ExecutorId::generate(),
1378 retry_config: ComponentRetryConfig::ZERO,
1379 locking_strategy,
1380 };
1381
1382 let reason = "error reason";
1383 let expected_reason = format!("activity trap: {reason}");
1384 let expected_detail = "error detail";
1385 let worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Err(
1386 WorkerError::ActivityTrap {
1387 reason: reason.to_string(),
1388 trap_kind: concepts::TrapKind::Trap,
1389 detail: Some(expected_detail.to_string()),
1390 version: Version::new(2),
1391 http_client_traces: None,
1392 },
1393 )));
1394 let execution_log = create_and_tick(
1395 CreateAndTickConfig {
1396 execution_id: ExecutionId::generate(),
1397 created_at,
1398 executed_at: created_at,
1399 },
1400 clock_fn,
1401 db_pool.connection_test().await.unwrap().as_ref(),
1402 db_pool.clone(),
1403 exec_config.clone(),
1404 worker,
1405 tick_fn,
1406 )
1407 .await;
1408 assert_eq!(3, execution_log.events.len());
1409 let (reason, kind, detail) = assert_matches!(
1410 &execution_log.events.get(2).unwrap(),
1411 ExecutionEvent {
1412 event: ExecutionRequest::Finished{
1413 retval: SupportedFunctionReturnValue::ExecutionError(FinishedExecutionError{reason, kind, detail}),
1414 http_client_traces: None
1415 },
1416 created_at: at,
1417 backtrace_id: None,
1418 version: Version(2),
1419 } if *at == created_at
1420 => (reason, kind, detail)
1421 );
1422
1423 assert_eq!(Some(expected_reason), *reason);
1424 assert_eq!(Some(expected_detail), detail.as_deref());
1425 assert_eq!(ExecutionFailureKind::Uncategorized, *kind);
1426
1427 db_close.close().await;
1428 }
1429
1430 #[rstest]
1431 #[tokio::test]
1432 async fn child_execution_permanently_failed_should_notify_parent_permanent_failure(
1433 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1434 locking_strategy: LockingStrategy,
1435 ) {
1436 let worker_error = WorkerError::ActivityTrap {
1437 reason: "error reason".to_string(),
1438 trap_kind: TrapKind::Trap,
1439 detail: Some("detail".to_string()),
1440 version: Version::new(2),
1441 http_client_traces: None,
1442 };
1443 let expected_child_err = FinishedExecutionError {
1444 kind: ExecutionFailureKind::Uncategorized,
1445 reason: Some("activity trap: error reason".to_string()),
1446 detail: Some("detail".to_string()),
1447 };
1448 child_execution_permanently_failed_should_notify_parent(
1449 WorkerResult::Err(worker_error),
1450 expected_child_err,
1451 locking_strategy,
1452 )
1453 .await;
1454 }
1455
1456 #[rstest]
1457 #[tokio::test]
1458 async fn child_execution_permanently_failed_handled_by_watcher_should_notify_parent_timeout(
1459 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1460 locking_strategy: LockingStrategy,
1461 ) {
1462 let expected_child_err = FinishedExecutionError {
1463 kind: ExecutionFailureKind::TimedOut,
1464 reason: None,
1465 detail: None,
1466 };
1467 child_execution_permanently_failed_should_notify_parent(
1468 WorkerResult::Ok(WorkerResultOk::DbUpdatedByWorkerOrWatcher),
1469 expected_child_err,
1470 locking_strategy,
1471 )
1472 .await;
1473 }
1474
1475 async fn child_execution_permanently_failed_should_notify_parent(
1476 worker_result: WorkerResult,
1477 expected_child_err: FinishedExecutionError,
1478 locking_strategy: LockingStrategy,
1479 ) {
1480 use concepts::storage::JoinSetResponseEventOuter;
1481 const LOCK_EXPIRY: Duration = Duration::from_secs(1);
1482
1483 set_up();
1484 let sim_clock = SimClock::default();
1485 let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1486
1487 let parent_worker = Arc::new(SimpleWorker::with_single_result(WorkerResult::Ok(
1488 WorkerResultOk::DbUpdatedByWorkerOrWatcher,
1489 )));
1490 let parent_execution_id = ExecutionId::generate();
1491 db_pool
1492 .connection()
1493 .await
1494 .unwrap()
1495 .create(CreateRequest {
1496 created_at: sim_clock.now(),
1497 execution_id: parent_execution_id.clone(),
1498 ffqn: FFQN_SOME,
1499 params: Params::empty(),
1500 parent: None,
1501 metadata: concepts::ExecutionMetadata::empty(),
1502 scheduled_at: sim_clock.now(),
1503 component_id: ComponentId::dummy_activity(),
1504 deployment_id: DEPLOYMENT_ID_DUMMY,
1505 scheduled_by: None,
1506 })
1507 .await
1508 .unwrap();
1509 let parent_executor_id = ExecutorId::generate();
1510 tick_fn(
1511 ExecConfig {
1512 batch_size: 1,
1513 lock_expiry: LOCK_EXPIRY,
1514 tick_sleep: Duration::ZERO,
1515 component_id: ComponentId::dummy_activity(),
1516 task_limiter: None,
1517 executor_id: parent_executor_id,
1518 retry_config: ComponentRetryConfig::ZERO,
1519 locking_strategy,
1520 },
1521 sim_clock.clone_box(),
1522 db_pool.clone(),
1523 parent_worker,
1524 sim_clock.now(),
1525 )
1526 .await;
1527
1528 let join_set_id = JoinSetId::new(JoinSetKind::OneOff, StrVariant::empty()).unwrap();
1529 let child_execution_id = parent_execution_id.next_level(&join_set_id);
1530 {
1532 let params = Params::empty();
1533 let child = CreateRequest {
1534 created_at: sim_clock.now(),
1535 execution_id: ExecutionId::Derived(child_execution_id.clone()),
1536 ffqn: FFQN_CHILD,
1537 params: params.clone(),
1538 parent: Some((parent_execution_id.clone(), join_set_id.clone())),
1539 metadata: concepts::ExecutionMetadata::empty(),
1540 scheduled_at: sim_clock.now(),
1541 component_id: ComponentId::dummy_activity(),
1542 deployment_id: DEPLOYMENT_ID_DUMMY,
1543 scheduled_by: None,
1544 };
1545 let current_time = sim_clock.now();
1546 let join_set = AppendRequest {
1547 created_at: current_time,
1548 event: ExecutionRequest::HistoryEvent {
1549 event: HistoryEvent::JoinSetCreate {
1550 join_set_id: join_set_id.clone(),
1551 },
1552 },
1553 };
1554 let child_exec_req = AppendRequest {
1555 created_at: current_time,
1556 event: ExecutionRequest::HistoryEvent {
1557 event: HistoryEvent::JoinSetRequest {
1558 join_set_id: join_set_id.clone(),
1559 request: JoinSetRequest::ChildExecutionRequest {
1560 child_execution_id: child_execution_id.clone(),
1561 target_ffqn: FFQN_CHILD,
1562 params,
1563 result: Ok(()),
1564 },
1565 },
1566 },
1567 };
1568 let join_next = AppendRequest {
1569 created_at: current_time,
1570 event: ExecutionRequest::HistoryEvent {
1571 event: HistoryEvent::JoinNext {
1572 join_set_id: join_set_id.clone(),
1573 run_expires_at: sim_clock.now(),
1574 closing: false,
1575 requested_ffqn: Some(FFQN_CHILD),
1576 },
1577 },
1578 };
1579 db_pool
1580 .connection()
1581 .await
1582 .unwrap()
1583 .append_batch_create_new_execution(
1584 current_time,
1585 vec![join_set, child_exec_req, join_next],
1586 parent_execution_id.clone(),
1587 Version::new(2),
1588 vec![child],
1589 vec![],
1590 )
1591 .await
1592 .unwrap();
1593 }
1594
1595 let child_worker =
1596 Arc::new(SimpleWorker::with_single_result(worker_result).with_ffqn(FFQN_CHILD));
1597
1598 tick_fn(
1600 ExecConfig {
1601 batch_size: 1,
1602 lock_expiry: LOCK_EXPIRY,
1603 tick_sleep: Duration::ZERO,
1604 component_id: ComponentId::dummy_activity(),
1605 task_limiter: None,
1606 executor_id: ExecutorId::generate(),
1607 retry_config: ComponentRetryConfig::ZERO,
1608 locking_strategy,
1609 },
1610 sim_clock.clone_box(),
1611 db_pool.clone(),
1612 child_worker,
1613 sim_clock.now(),
1614 )
1615 .await;
1616 if matches!(expected_child_err.kind, ExecutionFailureKind::TimedOut) {
1617 sim_clock.move_time_forward(LOCK_EXPIRY);
1619 expired_timers_watcher::tick(
1620 db_pool.connection().await.unwrap().as_ref(),
1621 sim_clock.now(),
1622 )
1623 .await
1624 .unwrap();
1625 }
1626 let child_log = db_pool
1627 .connection_test()
1628 .await
1629 .unwrap()
1630 .get(&ExecutionId::Derived(child_execution_id.clone()))
1631 .await
1632 .unwrap();
1633 assert!(child_log.pending_state.is_finished());
1634 assert_eq!(
1635 Version(2),
1636 child_log.next_version,
1637 "created = 0, locked = 1, with_single_result = 2"
1638 );
1639 assert_eq!(
1640 ExecutionRequest::Finished {
1641 retval: SupportedFunctionReturnValue::ExecutionError(expected_child_err),
1642 http_client_traces: None
1643 },
1644 child_log.last_event().event
1645 );
1646 let parent_log = db_pool
1647 .connection_test()
1648 .await
1649 .unwrap()
1650 .get(&parent_execution_id)
1651 .await
1652 .unwrap();
1653 assert_matches!(
1654 parent_log.pending_state,
1655 PendingState::PendingAt(PendingStatePendingAt {
1656 scheduled_at,
1657 last_lock: Some(LockedBy { executor_id: found_executor_id, run_id: _}),
1658 }) if scheduled_at == sim_clock.now() && found_executor_id == parent_executor_id,
1659 "parent should be back to pending"
1660 );
1661 let (found_join_set_id, found_child_execution_id, child_finished_version, found_result) = assert_matches!(
1662 parent_log.responses.last().map(|resp| &resp.event),
1663 Some(JoinSetResponseEventOuter{
1664 created_at: at,
1665 event: JoinSetResponseEvent{
1666 join_set_id: found_join_set_id,
1667 event: JoinSetResponse::ChildExecutionFinished {
1668 child_execution_id: found_child_execution_id,
1669 finished_version,
1670 result: found_result,
1671 }
1672 }
1673 })
1674 if *at == sim_clock.now()
1675 => (found_join_set_id, found_child_execution_id, finished_version, found_result)
1676 );
1677 assert_eq!(join_set_id, *found_join_set_id);
1678 assert_eq!(child_execution_id, *found_child_execution_id);
1679 assert_eq!(child_log.next_version, *child_finished_version);
1680 assert_matches!(
1681 found_result,
1682 SupportedFunctionReturnValue::ExecutionError(_)
1683 );
1684
1685 db_close.close().await;
1686 }
1687
1688 #[derive(Clone, Debug)]
1689 struct SleepyWorker {
1690 duration: Duration,
1691 result: SupportedFunctionReturnValue,
1692 exported: [FunctionMetadata; 1],
1693 }
1694
1695 #[async_trait]
1696 impl Worker for SleepyWorker {
1697 async fn run(&self, ctx: WorkerContext) -> WorkerResult {
1698 tokio::time::sleep(self.duration).await;
1699 WorkerResult::Ok(WorkerResultOk::RunFinished {
1700 retval: self.result.clone(),
1701 version: ctx.version,
1702 http_client_traces: None,
1703 })
1704 }
1705
1706 fn exported_functions_noext(&self) -> &[FunctionMetadata] {
1707 &self.exported
1708 }
1709 }
1710
1711 #[rstest]
1712 #[tokio::test]
1713 async fn hanging_lock_should_be_cleaned_and_execution_retried(
1714 #[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
1715 locking_strategy: LockingStrategy,
1716 ) {
1717 set_up();
1718 let sim_clock = SimClock::default();
1719 let (_guard, db_pool, db_close) = Database::Memory.set_up().await;
1720 let lock_expiry = Duration::from_millis(100);
1721 let timeout_duration = Duration::from_millis(300);
1722 let retry_config = ComponentRetryConfig {
1723 max_retries: Some(1),
1724 retry_exp_backoff: timeout_duration,
1725 };
1726 let exec_config = ExecConfig {
1727 batch_size: 1,
1728 lock_expiry,
1729 tick_sleep: Duration::ZERO,
1730 component_id: ComponentId::dummy_activity(),
1731 task_limiter: None,
1732 executor_id: ExecutorId::generate(),
1733 retry_config,
1734 locking_strategy,
1735 };
1736
1737 let worker = Arc::new(SleepyWorker {
1738 duration: lock_expiry + Duration::from_millis(1), result: SUPPORTED_RETURN_VALUE_OK_EMPTY,
1740 exported: [FunctionMetadata {
1741 ffqn: FFQN_SOME,
1742 parameter_types: ParameterTypes::default(),
1743 return_type: RETURN_TYPE_DUMMY,
1744 extension: None,
1745 submittable: true,
1746 }],
1747 });
1748 let execution_id = ExecutionId::generate();
1750 let db_connection = db_pool.connection_test().await.unwrap();
1751 db_connection
1752 .create(CreateRequest {
1753 created_at: sim_clock.now(),
1754 execution_id: execution_id.clone(),
1755 ffqn: FFQN_SOME,
1756 params: Params::empty(),
1757 parent: None,
1758 metadata: concepts::ExecutionMetadata::empty(),
1759 scheduled_at: sim_clock.now(),
1760 component_id: ComponentId::dummy_activity(),
1761 deployment_id: DEPLOYMENT_ID_DUMMY,
1762 scheduled_by: None,
1763 })
1764 .await
1765 .unwrap();
1766
1767 let ffqns = super::extract_exported_ffqns_noext(worker.as_ref());
1768 let executor = ExecTask::new_test(
1769 exec_config.clone(),
1770 worker,
1771 sim_clock.clone_box(),
1772 db_pool.clone(),
1773 ffqns,
1774 );
1775 let db_exec = db_pool.db_exec_conn().await.unwrap();
1776 let mut first_execution_progress = executor
1777 .tick(
1778 db_exec.as_ref(),
1779 sim_clock.now(),
1780 RunId::generate(),
1781 DEPLOYMENT_ID_DUMMY,
1782 )
1783 .await
1784 .unwrap();
1785 assert_eq!(1, first_execution_progress.executions.len());
1786 sim_clock.move_time_forward(lock_expiry);
1788 let now_after_first_lock_expiry = sim_clock.now();
1790 {
1791 debug!(now = %now_after_first_lock_expiry, "Expecting an expired lock");
1792 let cleanup_progress = executor
1793 .tick(
1794 db_pool.db_exec_conn().await.unwrap().as_ref(),
1795 now_after_first_lock_expiry,
1796 RunId::generate(),
1797 DEPLOYMENT_ID_DUMMY,
1798 )
1799 .await
1800 .unwrap();
1801 assert!(cleanup_progress.executions.is_empty());
1802 }
1803 {
1804 let expired_locks = expired_timers_watcher::tick(
1805 db_pool.connection().await.unwrap().as_ref(),
1806 now_after_first_lock_expiry,
1807 )
1808 .await
1809 .unwrap()
1810 .expired_locks;
1811 assert_eq!(1, expired_locks);
1812 }
1813 assert!(
1814 !first_execution_progress
1815 .executions
1816 .pop()
1817 .unwrap()
1818 .1
1819 .is_finished()
1820 );
1821
1822 let execution_log = db_connection.get(&execution_id).await.unwrap();
1823 let expected_first_timeout_expiry = now_after_first_lock_expiry + timeout_duration;
1824 assert_matches!(
1825 &execution_log.events.get(2).unwrap(),
1826 ExecutionEvent {
1827 event: ExecutionRequest::TemporarilyTimedOut { backoff_expires_at, .. },
1828 created_at: at,
1829 backtrace_id: None,
1830 version: Version(2),
1831 } if *at == now_after_first_lock_expiry && *backoff_expires_at == expected_first_timeout_expiry
1832 );
1833 assert_matches!(
1834 execution_log.pending_state,
1835 PendingState::PendingAt(PendingStatePendingAt {
1836 scheduled_at: found_scheduled_by,
1837 last_lock: Some(LockedBy {
1838 executor_id: found_executor_id,
1839 run_id: _,
1840 }),
1841 }) if found_scheduled_by == expected_first_timeout_expiry && found_executor_id == exec_config.executor_id
1842 );
1843 sim_clock.move_time_forward(timeout_duration);
1844 let now_after_first_timeout = sim_clock.now();
1845 debug!(now = %now_after_first_timeout, "Second execution should hang again and result in a permanent timeout");
1846
1847 let mut second_execution_progress = executor
1848 .tick(
1849 db_pool.db_exec_conn().await.unwrap().as_ref(),
1850 now_after_first_timeout,
1851 RunId::generate(),
1852 DEPLOYMENT_ID_DUMMY,
1853 )
1854 .await
1855 .unwrap();
1856 assert_eq!(1, second_execution_progress.executions.len());
1857
1858 sim_clock.move_time_forward(lock_expiry);
1860 let now_after_second_lock_expiry = sim_clock.now();
1862 debug!(now = %now_after_second_lock_expiry, "Expecting the second lock to be expired");
1863 {
1864 let cleanup_progress = executor
1865 .tick(
1866 db_pool.db_exec_conn().await.unwrap().as_ref(),
1867 now_after_second_lock_expiry,
1868 RunId::generate(),
1869 DEPLOYMENT_ID_DUMMY,
1870 )
1871 .await
1872 .unwrap();
1873 assert!(cleanup_progress.executions.is_empty());
1874 }
1875 {
1876 let expired_locks = expired_timers_watcher::tick(
1877 db_pool.connection().await.unwrap().as_ref(),
1878 now_after_second_lock_expiry,
1879 )
1880 .await
1881 .unwrap()
1882 .expired_locks;
1883 assert_eq!(1, expired_locks);
1884 }
1885 assert!(
1886 !second_execution_progress
1887 .executions
1888 .pop()
1889 .unwrap()
1890 .1
1891 .is_finished()
1892 );
1893
1894 drop(db_connection);
1895 drop(executor);
1896 db_close.close().await;
1897 }
1898}