1use std::collections::{BTreeSet, VecDeque};
2
3use runledger_core::jobs::{
4 WorkflowDependencyReleaseMode, WorkflowRunStatus, WorkflowStepExecutionKind, WorkflowStepStatus,
5};
6use serde_json::Value;
7use sqlx::types::Uuid;
8
9use crate::jobs::transaction_isolation::ensure_read_committed_tx;
10use crate::{DbTx, Error, Result};
11
12use super::super::row_decode::{
13 parse_job_stage, parse_job_type_name, parse_workflow_release_mode, parse_workflow_run_status,
14 parse_workflow_step_execution_kind, parse_workflow_step_status,
15};
16use super::super::rows::WorkflowStepRow;
17use super::super::workflow_types::{CompleteExternalWorkflowStepInput, WorkflowStepDbRecord};
18use super::errors::{
19 workflow_external_completion_conflict_error, workflow_external_completion_invalid_status_error,
20 workflow_external_completion_metadata_conflict_error,
21 workflow_external_completion_output_conflict_error,
22 workflow_external_completion_output_invalid_error, workflow_external_step_not_external_error,
23 workflow_external_step_not_found_error, workflow_external_step_not_waiting_error,
24 workflow_internal_state_error, workflow_release_conflict_error,
25};
26use super::locking::{
27 lock_workflow_run_release_shared_tx, lock_workflow_step_rows_for_update_tx,
28 try_lock_workflow_run_release_shared_tx,
29};
30use super::release::{StepReleaseCandidate, StepReleaseCandidateInit, release_candidate_step_tx};
31
32pub(crate) const WORKFLOW_RUN_TERMINAL_CHANNEL: &str = "runledger_workflow_run_terminal";
33
34fn validate_external_completion_status(terminal_status: WorkflowStepStatus) -> Result<()> {
35 if terminal_status.is_terminal() {
36 return Ok(());
37 }
38
39 Err(workflow_external_completion_invalid_status_error(
40 terminal_status,
41 ))
42}
43
44fn validate_external_completion_output(
45 terminal_status: WorkflowStepStatus,
46 output: Option<&Value>,
47) -> Result<()> {
48 if output.is_none() || terminal_status == WorkflowStepStatus::Succeeded {
49 return Ok(());
50 }
51
52 Err(workflow_external_completion_output_invalid_error(
53 terminal_status,
54 ))
55}
56
57fn validate_terminal_transition_status(terminal_status: WorkflowStepStatus) -> Result<()> {
58 if !terminal_status.is_terminal() {
59 return Err(workflow_internal_state_error(
60 "workflow step terminal transition requires terminal status",
61 ));
62 }
63
64 Ok(())
65}
66
67fn external_completion_metadata_matches(
68 step: &WorkflowStepDbRecord,
69 input: &CompleteExternalWorkflowStepInput<'_>,
70) -> bool {
71 step.status_reason.as_deref() == input.status_reason
72 && step.last_error_code.as_deref() == input.last_error_code
73 && step.last_error_message.as_deref() == input.last_error_message
74}
75
76async fn jsonb_values_match_tx(
77 tx: &mut DbTx<'_>,
78 left: Option<&Value>,
79 right: Option<&Value>,
80) -> Result<bool> {
81 let matches = sqlx::query_scalar!(
82 "SELECT $1::jsonb IS NOT DISTINCT FROM $2::jsonb AS \"matches!\"",
83 left,
84 right,
85 )
86 .fetch_one(&mut **tx)
87 .await
88 .map_err(|error| {
89 Error::from_query_sqlx_with_context("compare workflow external step output", error)
90 })?;
91
92 Ok(matches)
93}
94
95pub(crate) async fn mark_workflow_step_running_for_claim_tx(
96 tx: &mut DbTx<'_>,
97 job_id: Uuid,
98) -> Result<()> {
99 sqlx::query!(
100 "UPDATE workflow_steps
101 SET status = 'RUNNING',
102 started_at = COALESCE(started_at, now()),
103 output = NULL,
104 updated_at = now()
105 WHERE job_id = $1
106 AND status IN ('ENQUEUED', 'RUNNING')",
107 job_id,
108 )
109 .execute(&mut **tx)
110 .await
111 .map_err(|error| {
112 Error::from_query_sqlx_with_context("mark workflow step running for claim", error)
113 })?;
114
115 Ok(())
116}
117
118pub(crate) async fn mark_workflow_step_enqueued_for_claim_release_tx(
119 tx: &mut DbTx<'_>,
120 job_id: Uuid,
121 reset_started_at: bool,
122) -> Result<()> {
123 sqlx::query!(
124 "UPDATE workflow_steps
125 SET status = 'ENQUEUED',
126 started_at = CASE
127 WHEN $2 THEN NULL
128 ELSE started_at
129 END,
130 finished_at = NULL,
131 status_reason = NULL,
132 last_error_code = NULL,
133 last_error_message = NULL,
134 output = NULL,
135 updated_at = now()
136 WHERE job_id = $1
137 AND status = 'RUNNING'",
138 job_id,
139 reset_started_at,
140 )
141 .execute(&mut **tx)
142 .await
143 .map_err(|error| {
144 Error::from_query_sqlx_with_context("mark workflow step enqueued for claim release", error)
145 })?;
146
147 Ok(())
148}
149
150pub(crate) async fn mark_workflow_step_enqueued_for_retry_tx(
151 tx: &mut DbTx<'_>,
152 job_id: Uuid,
153 status_reason: Option<&str>,
154 last_error_code: Option<&str>,
155 last_error_message: Option<&str>,
156) -> Result<()> {
157 sqlx::query!(
163 "UPDATE workflow_steps
164 SET status = 'ENQUEUED',
165 finished_at = NULL,
166 status_reason = $2,
167 last_error_code = $3,
168 last_error_message = $4,
169 output = NULL,
170 updated_at = now()
171 WHERE job_id = $1
172 AND status = 'RUNNING'",
173 job_id,
174 status_reason,
175 last_error_code,
176 last_error_message,
177 )
178 .execute(&mut **tx)
179 .await
180 .map_err(|error| {
181 Error::from_query_sqlx_with_context("mark workflow step enqueued for retry", error)
182 })?;
183
184 Ok(())
185}
186
187pub(crate) async fn process_workflow_step_terminal_by_job_id_tx(
188 tx: &mut DbTx<'_>,
189 job_id: Uuid,
190 terminal_status: WorkflowStepStatus,
191 status_reason: Option<&str>,
192 last_error_code: Option<&str>,
193 last_error_message: Option<&str>,
194 output: Option<&Value>,
195) -> Result<()> {
196 validate_terminal_transition_status(terminal_status)?;
200
201 let linked_workflow_step_id: Option<Uuid> = sqlx::query_scalar!(
202 "SELECT workflow_step_id FROM job_queue WHERE id = $1 FOR UPDATE",
203 job_id
204 )
205 .fetch_optional(&mut **tx)
206 .await
207 .map_err(|error| {
208 Error::from_query_sqlx_with_context(
209 "lookup workflow step linkage for job integrity check",
210 error,
211 )
212 })?
213 .flatten();
214
215 if linked_workflow_step_id.is_some() {
216 ensure_read_committed_tx(
217 tx,
218 "workflow job terminal completion",
219 "workflow.terminal_completion_unsupported_isolation",
220 "Workflow job completion requires READ COMMITTED transaction isolation.",
221 )
222 .await?;
223 }
224
225 let row = sqlx::query!(
226 "SELECT id, workflow_run_id, status::text AS \"status!\"
227 FROM workflow_steps
228 WHERE job_id = $1
229 FOR UPDATE",
230 job_id,
231 )
232 .fetch_optional(&mut **tx)
233 .await
234 .map_err(|error| {
235 Error::from_query_sqlx_with_context(
236 "lock workflow step by job id for terminal update",
237 error,
238 )
239 })?;
240
241 let Some(row) = row else {
242 if let Some(workflow_step_id) = linked_workflow_step_id {
243 return Err(workflow_internal_state_error(format!(
244 "workflow-managed job {job_id} links to workflow step {workflow_step_id} but workflow_steps.job_id has no matching row"
245 )));
246 }
247
248 return Ok(());
249 };
250
251 let step_id: Uuid = row.id;
252 let workflow_run_id: Uuid = row.workflow_run_id;
253 let current_status = parse_workflow_step_status(row.status)?;
254
255 if linked_workflow_step_id != Some(step_id) {
256 let message = match linked_workflow_step_id {
257 Some(linked_step_id) => format!(
258 "workflow step linkage mismatch for job {job_id}: job_queue.workflow_step_id={linked_step_id}, workflow_steps.id={step_id}"
259 ),
260 None => format!(
261 "workflow step linkage mismatch for job {job_id}: job_queue.workflow_step_id is NULL, workflow_steps.id={step_id}"
262 ),
263 };
264 return Err(workflow_internal_state_error(message));
265 }
266
267 if current_status.is_terminal() {
268 return Ok(());
269 }
270
271 sqlx::query!(
272 "UPDATE workflow_steps
273 SET status = $2::text::workflow_step_status,
274 finished_at = COALESCE(finished_at, now()),
275 status_reason = $3,
276 last_error_code = $4,
277 last_error_message = $5,
278 output = CASE
279 WHEN $2::text::workflow_step_status = 'SUCCEEDED' THEN $6::jsonb
280 ELSE NULL
281 END,
282 updated_at = now()
283 WHERE id = $1",
284 step_id,
285 terminal_status.as_db_value(),
286 status_reason,
287 last_error_code,
288 last_error_message,
289 output,
290 )
291 .execute(&mut **tx)
292 .await
293 .map_err(|error| Error::from_query_sqlx_with_context("mark workflow step terminal", error))?;
294
295 let mut touched_run_ids = BTreeSet::from([workflow_run_id]);
296 lock_workflow_run_release_shared_tx(tx, workflow_run_id).await?;
308 resolve_terminal_step_queue_tx(tx, step_id, terminal_status, &mut touched_run_ids).await?;
309 recompute_workflow_run_statuses_tx(tx, &touched_run_ids).await?;
310
311 Ok(())
312}
313
314pub async fn complete_external_workflow_step(
315 pool: &crate::DbPool,
316 input: &CompleteExternalWorkflowStepInput<'_>,
317) -> Result<WorkflowStepDbRecord> {
318 let mut tx = pool
319 .begin()
320 .await
321 .map_err(|error| Error::ConnectionError(error.to_string()))?;
322 let step = complete_external_workflow_step_tx(&mut tx, input).await?;
323 tx.commit()
324 .await
325 .map_err(|error| Error::ConnectionError(error.to_string()))?;
326 Ok(step)
327}
328
329pub async fn complete_external_workflow_step_tx(
330 tx: &mut DbTx<'_>,
331 input: &CompleteExternalWorkflowStepInput<'_>,
332) -> Result<WorkflowStepDbRecord> {
333 validate_external_completion_status(input.terminal_status)?;
334 validate_external_completion_output(input.terminal_status, input.output)?;
335 ensure_read_committed_tx(
336 tx,
337 "workflow external step completion",
338 "workflow.external_completion_unsupported_isolation",
339 "External workflow step completion requires READ COMMITTED transaction isolation.",
340 )
341 .await?;
342 lock_workflow_step_rows_for_update_tx(tx, input.workflow_run_id, input.organization_id).await?;
343
344 let row = sqlx::query_as!(
345 WorkflowStepRow,
346 "SELECT
347 ws.id,
348 ws.workflow_run_id,
349 ws.step_key,
350 ws.execution_kind::text AS \"execution_kind!\",
351 ws.job_type,
352 ws.organization_id,
353 ws.payload,
354 ws.priority,
355 ws.max_attempts,
356 ws.timeout_seconds,
357 ws.stage,
358 ws.status::text AS \"status!\",
359 ws.job_id,
360 ws.released_at,
361 ws.started_at,
362 ws.finished_at,
363 ws.dependency_count_total,
364 ws.dependency_count_pending,
365 ws.dependency_count_unsatisfied,
366 ws.status_reason,
367 ws.last_error_code,
368 ws.last_error_message,
369 ws.output,
370 ws.created_at,
371 ws.updated_at
372 FROM workflow_steps ws
373 JOIN workflow_runs wr ON wr.id = ws.workflow_run_id
374 WHERE ws.workflow_run_id = $1
375 AND ws.step_key = $2
376 AND ($3::uuid IS NULL OR wr.organization_id = $3)
377 FOR UPDATE",
378 input.workflow_run_id,
379 input.step_key.as_str(),
380 input.organization_id,
381 )
382 .fetch_optional(&mut **tx)
383 .await
384 .map_err(|error| {
385 Error::from_query_sqlx_with_context("lock external workflow step for completion", error)
386 })?
387 .ok_or_else(workflow_external_step_not_found_error)?;
388
389 let stored_output = row.output.clone();
390 let step = row.into_record()?;
391 if step.execution_kind != WorkflowStepExecutionKind::External {
392 return Err(workflow_external_step_not_external_error(
393 step.step_key.as_str(),
394 ));
395 }
396
397 if step.status.is_terminal() {
398 if step.status == input.terminal_status {
399 if step.status == WorkflowStepStatus::Succeeded
400 && !jsonb_values_match_tx(tx, stored_output.as_ref(), input.output).await?
401 {
402 return Err(workflow_external_completion_output_conflict_error(
403 step.step_key.as_str(),
404 ));
405 }
406
407 if !external_completion_metadata_matches(&step, input) {
408 return Err(workflow_external_completion_metadata_conflict_error(
409 step.step_key.as_str(),
410 ));
411 }
412
413 return Ok(step);
414 }
415
416 return Err(workflow_external_completion_conflict_error(
417 step.step_key.as_str(),
418 step.status,
419 input.terminal_status,
420 ));
421 }
422
423 if step.status != WorkflowStepStatus::WaitingForExternal {
424 return Err(workflow_external_step_not_waiting_error(
425 step.step_key.as_str(),
426 step.status,
427 ));
428 }
429
430 if !try_lock_workflow_run_release_shared_tx(tx, step.workflow_run_id).await? {
431 return Err(workflow_release_conflict_error(step.workflow_run_id));
432 }
433
434 let updated = sqlx::query_as!(
435 WorkflowStepRow,
436 "UPDATE workflow_steps
437 SET status = $2::text::workflow_step_status,
438 finished_at = COALESCE(finished_at, now()),
439 status_reason = $3,
440 last_error_code = $4,
441 last_error_message = $5,
442 output = CASE
443 WHEN $2::text::workflow_step_status = 'SUCCEEDED' THEN $6::jsonb
444 ELSE NULL
445 END,
446 updated_at = now()
447 WHERE id = $1
448 RETURNING
449 id,
450 workflow_run_id,
451 step_key,
452 execution_kind::text AS \"execution_kind!\",
453 job_type,
454 organization_id,
455 payload,
456 priority,
457 max_attempts,
458 timeout_seconds,
459 stage,
460 status::text AS \"status!\",
461 job_id,
462 released_at,
463 started_at,
464 finished_at,
465 dependency_count_total,
466 dependency_count_pending,
467 dependency_count_unsatisfied,
468 status_reason,
469 last_error_code,
470 last_error_message,
471 output,
472 created_at,
473 updated_at",
474 step.id,
475 input.terminal_status.as_db_value(),
476 input.status_reason,
477 input.last_error_code,
478 input.last_error_message,
479 input.output,
480 )
481 .fetch_one(&mut **tx)
482 .await
483 .map_err(|error| {
484 Error::from_query_sqlx_with_context("mark external workflow step terminal", error)
485 })?;
486
487 let updated = updated.into_record()?;
488 let mut touched_run_ids = BTreeSet::from([updated.workflow_run_id]);
489 resolve_terminal_step_queue_tx(tx, updated.id, updated.status, &mut touched_run_ids).await?;
496 recompute_workflow_run_statuses_tx(tx, &touched_run_ids).await?;
497
498 Ok(updated)
499}
500
501pub(crate) async fn resolve_terminal_step_queue_tx(
502 tx: &mut DbTx<'_>,
503 initial_step_id: Uuid,
504 initial_terminal_status: WorkflowStepStatus,
505 touched_run_ids: &mut BTreeSet<Uuid>,
506) -> Result<()> {
507 let mut terminal_queue = VecDeque::from([(initial_step_id, initial_terminal_status)]);
508
509 while let Some((prerequisite_step_id, prerequisite_terminal_status)) =
510 terminal_queue.pop_front()
511 {
512 let edges = sqlx::query!(
513 "SELECT dependent_step_id, release_mode::text AS \"release_mode!\"
514 FROM workflow_step_dependencies
515 WHERE prerequisite_step_id = $1",
516 prerequisite_step_id,
517 )
518 .fetch_all(&mut **tx)
519 .await
520 .map_err(|error| {
521 Error::from_query_sqlx_with_context("lookup workflow step dependency edges", error)
522 })?;
523
524 for edge in edges {
525 let dependent_step_id: Uuid = edge.dependent_step_id;
526 let release_mode = parse_workflow_release_mode(edge.release_mode)?;
527 let dependency_unsatisfied =
528 matches!(release_mode, WorkflowDependencyReleaseMode::OnSuccess)
529 && !matches!(prerequisite_terminal_status, WorkflowStepStatus::Succeeded);
530
531 let row = sqlx::query!(
532 "UPDATE workflow_steps
533 SET dependency_count_pending = dependency_count_pending - 1,
534 dependency_count_unsatisfied = dependency_count_unsatisfied +
535 CASE WHEN $2 THEN 1 ELSE 0 END,
536 updated_at = now()
537 WHERE id = $1
538 RETURNING
539 id,
540 workflow_run_id,
541 execution_kind::text AS \"execution_kind!\",
542 job_type,
543 organization_id,
544 payload,
545 priority,
546 max_attempts,
547 timeout_seconds,
548 stage,
549 status::text AS \"status!\",
550 dependency_count_pending,
551 dependency_count_unsatisfied",
552 dependent_step_id,
553 dependency_unsatisfied,
554 )
555 .fetch_one(&mut **tx)
556 .await
557 .map_err(|error| {
558 Error::from_query_sqlx_with_context(
559 "update workflow step dependency counters",
560 error,
561 )
562 })?;
563
564 let candidate = StepReleaseCandidate::from_init(StepReleaseCandidateInit {
565 id: row.id,
566 workflow_run_id: row.workflow_run_id,
567 execution_kind: parse_workflow_step_execution_kind(row.execution_kind)?,
568 job_type: row.job_type.map(parse_job_type_name).transpose()?,
569 organization_id: row.organization_id,
570 payload: row.payload,
571 priority: row.priority,
572 max_attempts: row.max_attempts,
573 timeout_seconds: row.timeout_seconds,
574 stage: row.stage.map(parse_job_stage).transpose()?,
575 });
576 let status = parse_workflow_step_status(row.status)?;
577 let dependency_count_pending: i32 = row.dependency_count_pending;
578 let dependency_count_unsatisfied: i32 = row.dependency_count_unsatisfied;
579 touched_run_ids.insert(candidate.workflow_run_id());
580
581 if dependency_count_pending != 0 {
582 continue;
583 }
584
585 if status != WorkflowStepStatus::Blocked {
586 continue;
587 }
588
589 if dependency_count_unsatisfied == 0 {
590 release_candidate_step_tx(tx, &candidate).await?;
591 continue;
592 }
593
594 let canceled_row = sqlx::query_scalar!(
595 "UPDATE workflow_steps
596 SET status = 'CANCELED',
597 finished_at = COALESCE(finished_at, now()),
598 status_reason = 'workflow.dependency_unsatisfied',
599 last_error_code = 'workflow.dependency_unsatisfied',
600 last_error_message = 'Step dependency requirements were not satisfied.',
601 output = NULL,
602 updated_at = now()
603 WHERE id = $1
604 AND status = 'BLOCKED'
605 RETURNING workflow_run_id",
606 candidate.id(),
607 )
608 .fetch_optional(&mut **tx)
609 .await
610 .map_err(|error| {
611 Error::from_query_sqlx_with_context("cancel blocked workflow step", error)
612 })?;
613
614 if let Some(workflow_run_id) = canceled_row {
615 touched_run_ids.insert(workflow_run_id);
616 terminal_queue.push_back((candidate.id(), WorkflowStepStatus::Canceled));
617 }
618 }
619 }
620
621 Ok(())
622}
623
624pub(crate) async fn notify_workflow_run_terminal_tx(
625 tx: &mut DbTx<'_>,
626 workflow_run_id: Uuid,
627 status: WorkflowRunStatus,
628) -> Result<()> {
629 sqlx::query!(
630 "SELECT pg_notify(
631 $1,
632 json_build_object(
633 'workflow_run_id', $2::uuid::text,
634 'status', $3::text
635 )::text
636 )",
637 WORKFLOW_RUN_TERMINAL_CHANNEL,
638 workflow_run_id,
639 status.as_db_value(),
640 )
641 .execute(&mut **tx)
642 .await
643 .map_err(|error| Error::from_query_sqlx_with_context("notify workflow run terminal", error))?;
644
645 Ok(())
646}
647
648pub(crate) async fn recompute_workflow_run_statuses_tx(
649 tx: &mut DbTx<'_>,
650 touched_run_ids: &BTreeSet<Uuid>,
651) -> Result<()> {
652 for workflow_run_id in touched_run_ids {
653 let run_status = sqlx::query!(
656 "SELECT status::text AS \"status!\"
657 FROM workflow_runs
658 WHERE id = $1
659 FOR UPDATE",
660 *workflow_run_id
661 )
662 .fetch_optional(&mut **tx)
663 .await
664 .map_err(|error| {
665 Error::from_query_sqlx_with_context("lock workflow run for status recompute", error)
666 })?;
667 let Some(run_status) = run_status else {
668 continue;
669 };
670 let previous_status = parse_workflow_run_status(run_status.status)?;
671
672 let row = sqlx::query!(
673 "SELECT
674 COUNT(*) FILTER (
675 WHERE status IN ('BLOCKED', 'ENQUEUED', 'RUNNING')
676 )::bigint AS \"active_steps!\",
677 COUNT(*) FILTER (
678 WHERE status = 'WAITING_FOR_EXTERNAL'
679 )::bigint AS \"waiting_steps!\",
680 COUNT(*) FILTER (
681 WHERE status IN ('FAILED', 'CANCELED')
682 )::bigint AS \"errored_steps!\"
683 FROM workflow_steps
684 WHERE workflow_run_id = $1",
685 *workflow_run_id,
686 )
687 .fetch_one(&mut **tx)
688 .await
689 .map_err(|error| {
690 Error::from_query_sqlx_with_context("recompute workflow run status counters", error)
691 })?;
692
693 let active_steps: i64 = row.active_steps;
694 let waiting_steps: i64 = row.waiting_steps;
695 let errored_steps: i64 = row.errored_steps;
696
697 let next_status = if active_steps > 0 {
698 WorkflowRunStatus::Running
699 } else if waiting_steps > 0 {
700 WorkflowRunStatus::WaitingForExternal
701 } else if errored_steps > 0 {
702 WorkflowRunStatus::CompletedWithErrors
703 } else {
704 WorkflowRunStatus::Succeeded
705 };
706
707 if previous_status == next_status && next_status.is_terminal() {
708 continue;
709 }
710
711 let updated = sqlx::query!(
712 "UPDATE workflow_runs
713 SET status = $2::text::workflow_run_status,
714 finished_at = CASE
715 WHEN $2::text::workflow_run_status IN ('RUNNING', 'WAITING_FOR_EXTERNAL')
716 THEN NULL
717 ELSE COALESCE(finished_at, now())
718 END,
719 result = CASE
720 WHEN $2::text::workflow_run_status = 'SUCCEEDED' THEN (
721 SELECT ws.output
722 FROM workflow_steps ws
723 WHERE ws.workflow_run_id = workflow_runs.id
724 AND ws.step_key = workflow_runs.result_step_key
725 AND ws.status = 'SUCCEEDED'
726 LIMIT 1
727 )
728 WHEN $2::text::workflow_run_status IN ('COMPLETED_WITH_ERRORS', 'CANCELED')
729 THEN NULL
730 ELSE result
731 END,
732 updated_at = now()
733 WHERE id = $1
734 AND status <> 'CANCELED'",
735 *workflow_run_id,
736 next_status.as_db_value(),
737 )
738 .execute(&mut **tx)
739 .await
740 .map_err(|error| {
741 Error::from_query_sqlx_with_context("update workflow run recomputed status", error)
742 })?;
743
744 if updated.rows_affected() > 0
745 && next_status.is_terminal()
746 && !previous_status.is_terminal()
747 {
748 notify_workflow_run_terminal_tx(tx, *workflow_run_id, next_status).await?;
749 }
750 }
751
752 Ok(())
753}
754
755#[cfg(test)]
756mod tests {
757 use crate::{Error, QueryErrorCategory};
758
759 #[test]
760 fn workflow_terminal_transition_requires_terminal_status() {
761 let result = super::validate_terminal_transition_status(
762 runledger_core::jobs::WorkflowStepStatus::Running,
763 );
764 match result {
765 Err(Error::QueryError(query_error)) => {
766 assert_eq!(query_error.category(), QueryErrorCategory::Internal);
767 assert_eq!(query_error.code(), "workflow.internal_state");
768 assert!(
769 query_error
770 .internal_message()
771 .contains("workflow step terminal transition requires terminal status"),
772 "unexpected internal message: {}",
773 query_error.internal_message()
774 );
775 }
776 other => panic!("expected internal workflow state error, got {other:?}"),
777 }
778
779 assert!(
780 super::validate_terminal_transition_status(
781 runledger_core::jobs::WorkflowStepStatus::Succeeded
782 )
783 .is_ok()
784 );
785 assert!(
786 super::validate_terminal_transition_status(
787 runledger_core::jobs::WorkflowStepStatus::Failed
788 )
789 .is_ok()
790 );
791 }
792}