1use runledger_core::jobs::{JobStatus, JobType, WorkflowStepStatus};
2use serde_json::Value;
3use sqlx::types::Uuid;
4
5use crate::{DbPool, DbTx, Error, Result};
6
7use super::errors::{
8 invalid_job_state_error, job_not_found_error, validate_page_limit, validate_pagination,
9 workflow_requeue_not_supported_error,
10};
11use super::row_decode::{parse_job_event_type, parse_job_stage, parse_job_type_name};
12use super::rows::JobQueueRow;
13use super::types::{JobEventRecord, JobListFilter, JobMetricsRecord, JobQueueRecord};
14use super::workflows::on_terminal;
15
16const JOB_PAYLOAD_UUID_ARRAY_FIELD_UPDATE_LOCK_TIMEOUT: &str = "1s";
17const JOB_PAYLOAD_UUID_ARRAY_FIELD_UPDATE_LOCK_TIMEOUT_MS: i64 = 1_000;
18
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20#[must_use = "callers must inspect Updated/NotFound/Rejected"]
21#[non_exhaustive]
22pub enum JobPayloadUuidArrayFieldUpdate {
23 Updated,
24 NotFound,
25 Rejected {
26 reason: JobPayloadUuidArrayFieldUpdateRejection,
27 },
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31#[non_exhaustive]
32pub enum JobPayloadUuidArrayFieldUpdateRejection {
33 WorkflowManaged,
34 IdempotentRequestSnapshot,
35 NotPendingOrClaimed,
36}
37
38async fn rollback_and_classify_missing_job_mutation(
39 tx: DbTx<'_>,
40 pool: &DbPool,
41 organization_id: Option<Uuid>,
42 job_id: Uuid,
43) -> Result<Error> {
44 if let Err(error) = tx.rollback().await {
45 tracing::warn!(error = %error, "failed to rollback missing job mutation transaction");
46 }
47 let exists = get_job_by_id(pool, organization_id, job_id).await?;
48 Ok(if exists.is_none() {
49 job_not_found_error()
50 } else {
51 invalid_job_state_error()
52 })
53}
54
55async fn workflow_managed_job_exists_tx(
56 tx: &mut DbTx<'_>,
57 job_id: Uuid,
58 organization_id: Option<Uuid>,
59) -> Result<bool> {
60 let exists: bool = sqlx::query_scalar!(
61 "SELECT EXISTS (
62 SELECT 1
63 FROM job_queue jq
64 WHERE jq.id = $1
65 AND jq.workflow_step_id IS NOT NULL
66 AND ($2::uuid IS NULL OR jq.organization_id = $2)
67 ) AS \"exists!\"",
68 job_id,
69 organization_id,
70 )
71 .fetch_one(&mut **tx)
72 .await
73 .map_err(|error| {
74 Error::from_query_sqlx_with_context("requeue workflow-managed job check", error)
75 })?;
76
77 Ok(exists)
78}
79
80#[derive(sqlx::FromRow)]
81struct JobPayloadUuidArrayFieldUpdateCandidate {
82 status: String,
83 worker_id: Option<String>,
84 lease_expires_at: Option<chrono::DateTime<chrono::Utc>>,
85 workflow_step_id: Option<Uuid>,
86 idempotency_key: Option<String>,
87 enqueue_request: Option<Value>,
88}
89
90pub async fn list_jobs(pool: &DbPool, filter: &JobListFilter<'_>) -> Result<Vec<JobQueueRecord>> {
91 validate_pagination(filter.limit, filter.offset)?;
92
93 let status_filter = filter.status.map(JobStatus::as_db_value);
94
95 let rows = sqlx::query_as!(
96 JobQueueRow,
97 "SELECT
98 id,
99 job_type,
100 organization_id,
101 payload,
102 status::text AS \"status!\",
103 priority,
104 run_number,
105 attempt,
106 max_attempts,
107 timeout_seconds,
108 next_run_at,
109 lease_expires_at,
110 last_heartbeat_at,
111 worker_id,
112 started_at,
113 finished_at,
114 stage,
115 progress_done,
116 progress_total,
117 progress_pct::float8 AS progress_pct,
118 checkpoint,
119 output,
120 idempotency_key,
121 status_reason,
122 last_error_code,
123 last_error_message,
124 created_at,
125 updated_at
126 FROM job_queue
127 WHERE ($1::uuid IS NULL OR organization_id = $1)
128 AND ($2::text::job_status IS NULL OR status = $2::text::job_status)
129 AND ($3::text IS NULL OR job_type ILIKE '%' || $3 || '%')
130 ORDER BY created_at DESC, id DESC
131 LIMIT $4
132 OFFSET $5",
133 filter.organization_id,
134 status_filter,
135 filter.job_type,
136 filter.limit,
137 filter.offset,
138 )
139 .fetch_all(pool)
140 .await
141 .map_err(|error| Error::from_query_sqlx_with_context("list jobs", error))?;
142
143 rows.into_iter().map(JobQueueRow::into_record).collect()
144}
145
146pub async fn get_job_by_id(
147 pool: &DbPool,
148 organization_id: Option<Uuid>,
149 job_id: Uuid,
150) -> Result<Option<JobQueueRecord>> {
151 let row = sqlx::query_as!(
152 JobQueueRow,
153 "SELECT
154 id,
155 job_type,
156 organization_id,
157 payload,
158 status::text AS \"status!\",
159 priority,
160 run_number,
161 attempt,
162 max_attempts,
163 timeout_seconds,
164 next_run_at,
165 lease_expires_at,
166 last_heartbeat_at,
167 worker_id,
168 started_at,
169 finished_at,
170 stage,
171 progress_done,
172 progress_total,
173 progress_pct::float8 AS progress_pct,
174 checkpoint,
175 output,
176 idempotency_key,
177 status_reason,
178 last_error_code,
179 last_error_message,
180 created_at,
181 updated_at
182 FROM job_queue
183 WHERE id = $1
184 AND ($2::uuid IS NULL OR organization_id = $2)
185 LIMIT 1",
186 job_id,
187 organization_id,
188 )
189 .fetch_optional(pool)
190 .await
191 .map_err(|error| Error::from_query_sqlx_with_context("get job by id", error))?;
192
193 row.map(JobQueueRow::into_record).transpose()
194}
195
196pub async fn get_job_payload_by_idempotency_key(
197 pool: &DbPool,
198 organization_id: Uuid,
199 job_type: JobType<'_>,
200 idempotency_key: &str,
201) -> Result<Option<(Uuid, serde_json::Value)>> {
202 let row = sqlx::query!(
203 "SELECT id, payload
204 FROM job_queue
205 WHERE organization_id = $1
206 AND job_type = $2
207 AND idempotency_key = $3
208 LIMIT 1",
209 organization_id,
210 job_type as _,
211 idempotency_key,
212 )
213 .fetch_optional(pool)
214 .await
215 .map_err(|error| {
216 Error::from_query_sqlx_with_context("get job payload by idempotency key", error)
217 })?;
218
219 Ok(row.map(|row| (row.id, row.payload)))
220}
221
222pub async fn get_latest_job_payload_for_run(
223 pool: &DbPool,
224 organization_id: Uuid,
225 job_type: JobType<'_>,
226 run_id: Uuid,
227) -> Result<Option<(Uuid, serde_json::Value)>> {
228 let run_id_text = run_id.to_string();
229 let row = sqlx::query!(
230 "SELECT id, payload
231 FROM job_queue
232 WHERE organization_id = $1
233 AND job_type = $2
234 AND payload->>'run_id' = $3
235 ORDER BY created_at DESC, id DESC
236 LIMIT 1",
237 organization_id,
238 job_type as _,
239 run_id_text,
240 )
241 .fetch_optional(pool)
242 .await
243 .map_err(|error| {
244 Error::from_query_sqlx_with_context("get latest job payload for run", error)
245 })?;
246
247 Ok(row.map(|row| (row.id, row.payload)))
248}
249
250pub async fn update_job_payload_uuid_array_field(
256 pool: &DbPool,
257 organization_id: Uuid,
258 job_id: Uuid,
259 job_type: JobType<'_>,
260 payload_field: &str,
261 values: &[Uuid],
262) -> Result<JobPayloadUuidArrayFieldUpdate> {
263 let mut tx = pool.begin().await.map_err(|error| {
264 Error::from_query_sqlx_with_context(
265 "begin job payload uuid array update transaction",
266 error,
267 )
268 })?;
269
270 let previous_lock_timeout =
271 cap_job_payload_uuid_array_field_update_lock_timeout_tx(&mut tx).await?;
272
273 let row_result = sqlx::query_as::<_, JobPayloadUuidArrayFieldUpdateCandidate>(
274 "SELECT
275 status::text AS status,
276 worker_id,
277 lease_expires_at,
278 workflow_step_id,
279 idempotency_key,
280 enqueue_request
281 FROM job_queue
282 WHERE id = $1
283 AND organization_id = $2
284 AND job_type = $3
285 FOR UPDATE",
286 )
287 .bind(job_id)
288 .bind(organization_id)
289 .bind(job_type)
290 .fetch_optional(&mut *tx)
291 .await;
292
293 let row = match row_result {
294 Ok(row) => {
295 set_local_lock_timeout_tx(
296 &mut tx,
297 &previous_lock_timeout,
298 "restore job payload uuid array update lock timeout",
299 )
300 .await?;
301 row
302 }
303 Err(error) => {
304 return Err(Error::from_query_sqlx_with_context(
305 "classify job payload uuid array update",
306 error,
307 ));
308 }
309 };
310
311 let Some(row) = row else {
312 tx.commit().await.map_err(|error| {
313 Error::from_query_sqlx_with_context(
314 "commit job payload uuid array update transaction",
315 error,
316 )
317 })?;
318 return Ok(JobPayloadUuidArrayFieldUpdate::NotFound);
319 };
320
321 let rejection = if row.workflow_step_id.is_some() {
324 Some(JobPayloadUuidArrayFieldUpdateRejection::WorkflowManaged)
325 } else if row.idempotency_key.is_some() || row.enqueue_request.is_some() {
326 Some(JobPayloadUuidArrayFieldUpdateRejection::IdempotentRequestSnapshot)
327 } else if row.status != JobStatus::Pending.as_db_value()
328 || row.worker_id.is_some()
329 || row.lease_expires_at.is_some()
330 {
331 Some(JobPayloadUuidArrayFieldUpdateRejection::NotPendingOrClaimed)
332 } else {
333 None
334 };
335
336 if let Some(reason) = rejection {
337 tx.commit().await.map_err(|error| {
338 Error::from_query_sqlx_with_context(
339 "commit job payload uuid array update transaction",
340 error,
341 )
342 })?;
343 return Ok(JobPayloadUuidArrayFieldUpdate::Rejected { reason });
344 }
345
346 sqlx::query!(
347 "UPDATE job_queue
348 SET
349 payload = jsonb_set(
350 payload,
351 ARRAY[$4::text],
352 to_jsonb($5::uuid[]),
353 true
354 ),
355 updated_at = now()
356 WHERE id = $1
357 AND organization_id = $2
358 AND job_type = $3",
359 job_id,
360 organization_id,
361 job_type as _,
362 payload_field,
363 values,
364 )
365 .execute(&mut *tx)
366 .await
367 .map_err(|error| {
368 Error::from_query_sqlx_with_context("update job payload uuid array field", error)
369 })?;
370
371 tx.commit().await.map_err(|error| {
372 Error::from_query_sqlx_with_context(
373 "commit job payload uuid array update transaction",
374 error,
375 )
376 })?;
377 Ok(JobPayloadUuidArrayFieldUpdate::Updated)
378}
379
380async fn cap_job_payload_uuid_array_field_update_lock_timeout_tx(
381 tx: &mut DbTx<'_>,
382) -> Result<String> {
383 sqlx::query_scalar::<_, String>(
384 "WITH previous AS MATERIALIZED (
385 SELECT
386 current_setting('lock_timeout') AS lock_timeout,
387 setting::bigint AS lock_timeout_ms
388 FROM pg_settings
389 WHERE name = 'lock_timeout'
390 )
391 SELECT previous.lock_timeout
392 FROM previous,
393 LATERAL (
394 SELECT set_config(
395 'lock_timeout',
396 CASE
397 WHEN previous.lock_timeout_ms = 0 THEN $1
398 WHEN previous.lock_timeout_ms <= $2 THEN previous.lock_timeout
399 ELSE $1
400 END,
401 true
402 )
403 ) AS applied",
404 )
405 .bind(JOB_PAYLOAD_UUID_ARRAY_FIELD_UPDATE_LOCK_TIMEOUT)
406 .bind(JOB_PAYLOAD_UUID_ARRAY_FIELD_UPDATE_LOCK_TIMEOUT_MS)
407 .fetch_one(&mut **tx)
408 .await
409 .map_err(|error| {
410 Error::from_query_sqlx_with_context("set job payload uuid array update lock timeout", error)
411 })
412}
413
414async fn set_local_lock_timeout_tx(
415 tx: &mut DbTx<'_>,
416 lock_timeout: &str,
417 context: &'static str,
418) -> Result<()> {
419 sqlx::query_scalar::<_, String>("SELECT set_config('lock_timeout', $1, true)")
420 .bind(lock_timeout)
421 .fetch_one(&mut **tx)
422 .await
423 .map_err(|error| Error::from_query_sqlx_with_context(context, error))?;
424
425 Ok(())
426}
427
428pub async fn list_job_events(
429 pool: &DbPool,
430 organization_id: Option<Uuid>,
431 job_id: Uuid,
432 limit: i64,
433 after_id: Option<i64>,
434) -> Result<Vec<JobEventRecord>> {
435 validate_page_limit(limit)?;
436
437 let rows = sqlx::query!(
438 "SELECT
439 je.id,
440 je.job_id,
441 je.run_number,
442 je.attempt,
443 je.event_type::text AS \"event_type!\",
444 je.stage,
445 je.progress_done,
446 je.progress_total,
447 je.payload,
448 je.occurred_at
449 FROM job_events je
450 JOIN job_queue jq ON jq.id = je.job_id
451 WHERE je.job_id = $1
452 AND ($2::uuid IS NULL OR jq.organization_id = $2)
453 AND ($3::bigint IS NULL OR je.id > $3)
454 ORDER BY je.id ASC
455 LIMIT $4",
456 job_id,
457 organization_id,
458 after_id,
459 limit,
460 )
461 .fetch_all(pool)
462 .await
463 .map_err(|error| Error::from_query_sqlx_with_context("list job events", error))?;
464
465 rows.into_iter()
466 .map(|row| {
467 Ok(JobEventRecord {
468 id: row.id,
469 job_id: row.job_id,
470 run_number: row.run_number,
471 attempt: row.attempt,
472 event_type: parse_job_event_type(row.event_type)?,
473 stage: row.stage.map(parse_job_stage).transpose()?,
474 progress_done: row.progress_done,
475 progress_total: row.progress_total,
476 payload: row.payload,
477 occurred_at: row.occurred_at,
478 })
479 })
480 .collect::<Result<Vec<_>>>()
481}
482
483pub async fn get_job_metrics(
484 pool: &DbPool,
485 organization_id: Option<Uuid>,
486 job_type: Option<&str>,
487) -> Result<Vec<JobMetricsRecord>> {
488 let rows = sqlx::query!(
489 "SELECT
490 jd.job_type AS \"job_type!\",
491 COALESCE(SUM(jmr.pending_count), 0)::bigint AS \"pending_count!\",
492 COALESCE(SUM(jmr.leased_count), 0)::bigint AS \"leased_count!\",
493 COALESCE(SUM(jmr.stale_leases), 0)::bigint AS \"stale_leases!\",
494 COALESCE(SUM(jmr.succeeded_24h), 0)::bigint AS \"succeeded_24h!\",
495 COALESCE(SUM(jmr.retryable_24h), 0)::bigint AS \"retryable_24h!\",
496 COALESCE(SUM(jmr.terminal_24h), 0)::bigint AS \"terminal_24h!\",
497 COALESCE(SUM(jmr.panicked_24h), 0)::bigint AS \"panicked_24h!\",
498 COALESCE(SUM(jmr.timeout_24h), 0)::bigint AS \"timeout_24h!\",
499 COALESCE(SUM(jmr.dead_lettered_24h), 0)::bigint AS \"dead_lettered_24h!\",
500 AVG(jmr.p50_duration_ms_24h) AS p50_duration_ms_24h,
501 AVG(jmr.p95_duration_ms_24h) AS p95_duration_ms_24h
502 FROM job_definitions jd
503 LEFT JOIN job_metrics_rollup jmr
504 ON jmr.job_type = jd.job_type
505 AND ($1::uuid IS NULL OR jmr.organization_id = $1)
506 WHERE ($2::text IS NULL OR jd.job_type = $2)
507 GROUP BY jd.job_type
508 ORDER BY jd.job_type ASC",
509 organization_id,
510 job_type,
511 )
512 .fetch_all(pool)
513 .await
514 .map_err(|error| Error::from_query_sqlx_with_context("get job metrics", error))?;
515
516 rows.into_iter()
517 .map(|row| {
518 Ok(JobMetricsRecord {
519 job_type: parse_job_type_name(row.job_type)?,
520 pending_count: row.pending_count,
521 leased_count: row.leased_count,
522 stale_leases: row.stale_leases,
523 succeeded_24h: row.succeeded_24h,
524 retryable_24h: row.retryable_24h,
525 terminal_24h: row.terminal_24h,
526 panicked_24h: row.panicked_24h,
527 timeout_24h: row.timeout_24h,
528 dead_lettered_24h: row.dead_lettered_24h,
529 p50_duration_ms_24h: row.p50_duration_ms_24h,
530 p95_duration_ms_24h: row.p95_duration_ms_24h,
531 })
532 })
533 .collect::<Result<Vec<_>>>()
534}
535
536pub async fn cancel_job(
537 pool: &DbPool,
538 organization_id: Option<Uuid>,
539 job_id: Uuid,
540 reason: Option<&str>,
541) -> Result<JobQueueRecord> {
542 let mut tx = pool
543 .begin()
544 .await
545 .map_err(|error| Error::ConnectionError(error.to_string()))?;
546 let Some(record) = cancel_job_tx(&mut tx, organization_id, job_id, reason).await? else {
547 return Err(
548 rollback_and_classify_missing_job_mutation(tx, pool, organization_id, job_id).await?,
549 );
550 };
551
552 tx.commit()
553 .await
554 .map_err(|error| Error::ConnectionError(error.to_string()))?;
555
556 Ok(record)
557}
558
559pub(crate) async fn cancel_job_tx(
560 tx: &mut DbTx<'_>,
561 organization_id: Option<Uuid>,
562 job_id: Uuid,
563 reason: Option<&str>,
564) -> Result<Option<JobQueueRecord>> {
565 let row = sqlx::query_as!(
566 JobQueueRow,
567 "UPDATE job_queue
568 SET status = 'CANCELED',
569 lease_expires_at = NULL,
570 last_heartbeat_at = NULL,
571 worker_id = NULL,
572 finished_at = now(),
573 output = NULL,
574 status_reason = COALESCE($3, 'CANCELED'),
575 updated_at = now()
576 WHERE id = $1
577 AND ($2::uuid IS NULL OR organization_id = $2)
578 AND status IN ('PENDING', 'LEASED')
579 RETURNING
580 id,
581 job_type,
582 organization_id,
583 payload,
584 status::text AS \"status!\",
585 priority,
586 run_number,
587 attempt,
588 max_attempts,
589 timeout_seconds,
590 next_run_at,
591 lease_expires_at,
592 last_heartbeat_at,
593 worker_id,
594 started_at,
595 finished_at,
596 stage,
597 progress_done,
598 progress_total,
599 progress_pct::float8 AS progress_pct,
600 checkpoint,
601 output,
602 idempotency_key,
603 status_reason,
604 last_error_code,
605 last_error_message,
606 created_at,
607 updated_at",
608 job_id,
609 organization_id,
610 reason,
611 )
612 .fetch_optional(&mut **tx)
613 .await
614 .map_err(|error| Error::from_query_sqlx_with_context("cancel job", error))?;
615
616 let Some(row) = row else {
617 return Ok(None);
618 };
619
620 let record = row.into_record()?;
621
622 sqlx::query!(
623 "UPDATE job_attempts
624 SET finished_at = now()
625 WHERE job_id = $1
626 AND run_number = $2
627 AND attempt = $3
628 AND finished_at IS NULL",
629 record.id,
630 record.run_number,
631 record.attempt,
632 )
633 .execute(&mut **tx)
634 .await
635 .map_err(|error| Error::from_query_sqlx_with_context("close canceled attempt", error))?;
636
637 let event_attempt = (record.attempt > 0).then_some(record.attempt);
638 sqlx::query!(
639 "INSERT INTO job_events (
640 job_id,
641 run_number,
642 attempt,
643 event_type,
644 payload
645 )
646 VALUES (
647 $1,
648 $2,
649 $3,
650 'CANCELED',
651 jsonb_build_object('reason', $4::text)
652 )",
653 record.id,
654 record.run_number,
655 event_attempt,
656 record.status_reason.as_deref(),
657 )
658 .execute(&mut **tx)
659 .await
660 .map_err(|error| Error::from_query_sqlx_with_context("insert canceled event", error))?;
661
662 on_terminal(
663 tx,
664 record.id,
665 WorkflowStepStatus::Canceled,
666 record.status_reason.as_deref(),
667 None,
668 None,
669 None,
670 )
671 .await?;
672
673 Ok(Some(record))
674}
675
676fn ensure_workflow_requeue_rejection_rollback_succeeded(
677 rollback_result: std::result::Result<(), sqlx::Error>,
678) -> Result<()> {
679 rollback_result.map_err(|error| Error::ConnectionError(error.to_string()))
680}
681
682pub async fn requeue_job(
683 pool: &DbPool,
684 organization_id: Option<Uuid>,
685 job_id: Uuid,
686 reason: Option<&str>,
687) -> Result<JobQueueRecord> {
688 let mut tx = pool
689 .begin()
690 .await
691 .map_err(|error| Error::ConnectionError(error.to_string()))?;
692
693 let workflow_managed = workflow_managed_job_exists_tx(&mut tx, job_id, organization_id).await?;
694 if workflow_managed {
695 ensure_workflow_requeue_rejection_rollback_succeeded(tx.rollback().await)?;
696 return Err(workflow_requeue_not_supported_error());
697 }
698
699 let previous_run = sqlx::query!(
700 "SELECT run_number, attempt
701 FROM job_queue
702 WHERE id = $1
703 AND ($2::uuid IS NULL OR organization_id = $2)
704 AND status IN ('DEAD_LETTERED', 'CANCELED', 'SUCCEEDED')
705 FOR UPDATE",
706 job_id,
707 organization_id,
708 )
709 .fetch_optional(&mut *tx)
710 .await
711 .map_err(|error| Error::from_query_sqlx_with_context("requeue job prefetch attempt", error))?;
712
713 let Some(previous_run) = previous_run else {
714 return Err(
715 rollback_and_classify_missing_job_mutation(tx, pool, organization_id, job_id).await?,
716 );
717 };
718 let previous_run_number: i32 = previous_run.run_number;
719 let previous_attempt: i32 = previous_run.attempt;
720
721 let row = sqlx::query_as!(
722 JobQueueRow,
723 "UPDATE job_queue
724 SET status = 'PENDING',
725 stage = 'queued',
726 progress_done = NULL,
727 progress_total = NULL,
728 checkpoint = NULL,
729 output = NULL,
730 run_number = run_number + 1,
731 attempt = 0,
732 lease_expires_at = NULL,
733 last_heartbeat_at = NULL,
734 worker_id = NULL,
735 next_run_at = now(),
736 started_at = NULL,
737 finished_at = NULL,
738 status_reason = COALESCE($3, 'REQUEUED'),
739 last_error_code = NULL,
740 last_error_message = NULL,
741 updated_at = now()
742 WHERE id = $1
743 AND ($2::uuid IS NULL OR organization_id = $2)
744 AND status IN ('DEAD_LETTERED', 'CANCELED', 'SUCCEEDED')
745 RETURNING
746 id,
747 job_type,
748 organization_id,
749 payload,
750 status::text AS \"status!\",
751 priority,
752 run_number,
753 attempt,
754 max_attempts,
755 timeout_seconds,
756 next_run_at,
757 lease_expires_at,
758 last_heartbeat_at,
759 worker_id,
760 started_at,
761 finished_at,
762 stage,
763 progress_done,
764 progress_total,
765 progress_pct::float8 AS progress_pct,
766 checkpoint,
767 output,
768 idempotency_key,
769 status_reason,
770 last_error_code,
771 last_error_message,
772 created_at,
773 updated_at",
774 job_id,
775 organization_id,
776 reason,
777 )
778 .fetch_optional(&mut *tx)
779 .await
780 .map_err(|error| Error::from_query_sqlx_with_context("requeue job", error))?;
781
782 let Some(row) = row else {
783 return Err(
784 rollback_and_classify_missing_job_mutation(tx, pool, organization_id, job_id).await?,
785 );
786 };
787
788 let record = row.into_record()?;
789
790 let event_attempt = (previous_attempt > 0).then_some(previous_attempt);
791 sqlx::query!(
792 "INSERT INTO job_events (
793 job_id,
794 run_number,
795 attempt,
796 event_type,
797 payload
798 )
799 VALUES (
800 $1,
801 $2,
802 $3,
803 'REQUEUED',
804 jsonb_build_object('reason', $4::text)
805 )",
806 record.id,
807 previous_run_number,
808 event_attempt,
809 record.status_reason.as_deref(),
810 )
811 .execute(&mut *tx)
812 .await
813 .map_err(|error| Error::from_query_sqlx_with_context("insert requeued event", error))?;
814
815 tx.commit()
816 .await
817 .map_err(|error| Error::ConnectionError(error.to_string()))?;
818
819 Ok(record)
820}
821
822#[cfg(test)]
823mod tests {
824 use super::ensure_workflow_requeue_rejection_rollback_succeeded;
825 use crate::Error;
826
827 #[test]
828 fn workflow_requeue_rejection_allows_validation_error_after_successful_rollback() {
829 let result = ensure_workflow_requeue_rejection_rollback_succeeded(Ok(()));
830 assert!(result.is_ok());
831 }
832
833 #[test]
834 fn workflow_requeue_rejection_returns_internal_error_when_rollback_fails() {
835 let result =
836 ensure_workflow_requeue_rejection_rollback_succeeded(Err(sqlx::Error::PoolTimedOut));
837 match result {
838 Err(Error::ConnectionError(message)) => assert!(!message.is_empty()),
839 other => panic!("expected connection error, got {other:?}"),
840 }
841 }
842}