Skip to content

Commit ca3f456

Browse files
rmcdanielclaude
andcommitted
Rust #325: one deadline close for completions, failures and history waits
Completions and failure reports both checked the run's deadlines and closed the run with the same timeout, commit and receipt. They now share close_expired_run, which hands the transaction back while the run is inside its deadlines (Sonar flagged the copy on #926). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 395b50c commit ca3f456

1 file changed

Lines changed: 32 additions & 26 deletions

File tree

‎rust/src/runtime/store.rs‎

Lines changed: 32 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -2605,6 +2605,34 @@ where
26052605
Ok(response)
26062606
}
26072607

2608+
/// `timeoutIfDeadlineExpired`: a workflow task report on a run whose
2609+
/// execution or run deadline has passed closes the run as timed out and
2610+
/// refuses the report with the completion's receipt (`409`, outcome
2611+
/// `completed`, `run_status` `failed`, reason `run_timed_out`). PHP SDK
2612+
/// 3.6.1 recognises that receipt, and does not recognise a timeout refusal
2613+
/// shaped as `waiting_for_history` (Server #794). Completions, failures
2614+
/// and history waits all use it. A run still inside its deadlines gets its
2615+
/// transaction back unchanged.
2616+
async fn close_expired_run(
2617+
&self,
2618+
mut tx: Transaction<'static, DB>,
2619+
run: &DB::Row,
2620+
task: &DB::Row,
2621+
attempt: &Value,
2622+
) -> Result<Transaction<'static, DB>> {
2623+
let expired = [DB::optional_instant(run, "execution_deadline_at")?, DB::optional_instant(run, "run_deadline_at")?]
2624+
.into_iter()
2625+
.flatten()
2626+
.any(|deadline| deadline <= now());
2627+
if !expired {
2628+
return Ok(tx);
2629+
}
2630+
let response = Self::time_out_run(&mut tx, run, task, attempt).await?;
2631+
tx.commit().await?;
2632+
self.wake.notify_waiters();
2633+
Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response })
2634+
}
2635+
26082636
pub(crate) async fn fail_workflow_task(
26092637
&self,
26102638
task_id: &str,
@@ -2692,22 +2720,9 @@ where
26922720
let run_id = DB::string(&task, "workflow_run_id")?;
26932721
let run = Self::active_run(&mut tx, &run_id).await?;
26942722
// Server #794: a failure or history wait reported after the run's
2695-
// deadline closes the run as timed out, and the worker gets the same
2696-
// receipt a completion would (outcome `completed`, `run_status`
2697-
// `failed`, reason `run_timed_out`): PHP SDK 3.6.1 recognises that
2698-
// receipt, and does not recognise a timeout refusal shaped as
2699-
// `waiting_for_history`. Published 2.11.8
2700-
// acknowledged the wait and left the run waiting.
2701-
if [DB::optional_instant(&run, "execution_deadline_at")?, DB::optional_instant(&run, "run_deadline_at")?]
2702-
.into_iter()
2703-
.flatten()
2704-
.any(|deadline| deadline <= now())
2705-
{
2706-
let response = Self::time_out_run(&mut tx, &run, &task, &body["workflow_task_attempt"]).await?;
2707-
tx.commit().await?;
2708-
self.wake.notify_waiters();
2709-
return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response });
2710-
}
2723+
// deadline closes the run as timed out, with the receipt a completion
2724+
// gets. Published 2.11.8 acknowledged the wait and left the run waiting.
2725+
let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?;
27112726
let mut payload = DB::document_row(&task, "payload")?;
27122727
// Published PHP retries a failed task as it is, with any signal or
27132728
// update it carries. A waiting acknowledgement of a signal task
@@ -3107,16 +3122,7 @@ where
31073122
let run = Self::active_run(&mut tx, &run_id).await?;
31083123
// `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines
31093124
// when one of its workflow tasks completes, and closes the run.
3110-
if [DB::optional_instant(&run, "execution_deadline_at")?, DB::optional_instant(&run, "run_deadline_at")?]
3111-
.into_iter()
3112-
.flatten()
3113-
.any(|deadline| deadline <= now())
3114-
{
3115-
let response = Self::time_out_run(&mut tx, &run, &task, &body["workflow_task_attempt"]).await?;
3116-
tx.commit().await?;
3117-
self.wake.notify_waiters();
3118-
return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response });
3119-
}
3125+
let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?;
31203126
// Workflow #930: the deadline close above is never metered. Before
31213127
// the worker's first write, its local activities' exact events and
31223128
// rows (other commands at one event) must fit the limits; then the

0 commit comments

Comments
 (0)