diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 85cd9aefd..28e38ed27 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -466,6 +466,27 @@ contract fixture before its correction. task, oldest first, as PHP does behind a leased task. PHP's merge is reported as [server#902](https://github.com/durable-workflow/server/issues/902). +- **A failure or history wait reported after the run's deadline** + (November 1, decided: native differs from published 2.11.8 on purpose). + Published 2.11.8 acknowledges a `WorkflowTaskWaitingForHistory` failure + after the run or execution deadline and leaves the run `waiting`, so a + deadline probe never closes it + ([server#794 finding](https://github.com/durable-workflow/server/issues/794#issuecomment-6562291610)). + Native closes the run when a workflow task failure, waiting or not, is + reported after either deadline, through the same timeout a completion + uses, as General's correction does (server `c2b18fad` and its follow-up + on #794, not yet published). The worker gets the completion's receipt + for a timed-out run: `409`, `outcome` `completed`, `recorded` false, + `run_status` `failed`, `reason` `run_timed_out`, with the original + `task_id`, `workflow_task_attempt` and `run_id`. PHP SDK 3.6.1 recognises + that receipt, and does not recognise a timeout refusal shaped as + `waiting_for_history`. The run gets one + `WorkflowTimedOut` with the deadline's `timeout_kind`, and no workflow + task stays open. A stale attempt still changes nothing, and a replay adds + no second timeout + (`run_deadlines::reports_after_a_deadline_time_out_the_run_with_the_completion_receipt`). + Record a parity fixture once a PHP release with the correction is + published. - **Timer firing under the CPU deadline profile on slower hosts.** In local MySQL 8.0 cells, where eight dense readers saturate the one-CPU database and the one-CPU native server, one of six runs fired the three-second timer diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index f58159032..af22d6c81 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2605,6 +2605,34 @@ where Ok(response) } + /// `timeoutIfDeadlineExpired`: a workflow task report on a run whose + /// execution or run deadline has passed closes the run as timed out and + /// refuses the report with the completion's receipt (`409`, outcome + /// `completed`, `run_status` `failed`, reason `run_timed_out`). PHP SDK + /// 3.6.1 recognises that receipt, and does not recognise a timeout refusal + /// shaped as `waiting_for_history` (Server #794). Completions, failures + /// and history waits all use it. A run still inside its deadlines gets its + /// transaction back unchanged. + async fn close_expired_run( + &self, + mut tx: Transaction<'static, DB>, + run: &DB::Row, + task: &DB::Row, + attempt: &Value, + ) -> Result> { + let expired = [DB::optional_instant(run, "execution_deadline_at")?, DB::optional_instant(run, "run_deadline_at")?] + .into_iter() + .flatten() + .any(|deadline| deadline <= now()); + if !expired { + return Ok(tx); + } + let response = Self::time_out_run(&mut tx, run, task, attempt).await?; + tx.commit().await?; + self.wake.notify_waiters(); + Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response }) + } + pub(crate) async fn fail_workflow_task( &self, task_id: &str, @@ -2691,6 +2719,10 @@ where Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds).await?; let run_id = DB::string(&task, "workflow_run_id")?; let run = Self::active_run(&mut tx, &run_id).await?; + // Server #794: a failure or history wait reported after the run's + // deadline closes the run as timed out, with the receipt a completion + // gets. Published 2.11.8 acknowledged the wait and left the run waiting. + let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?; let mut payload = DB::document_row(&task, "payload")?; // Published PHP retries a failed task as it is, with any signal or // update it carries. A waiting acknowledgement of a signal task @@ -3090,16 +3122,7 @@ where let run = Self::active_run(&mut tx, &run_id).await?; // `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines // when one of its workflow tasks completes, and closes the run. - if [DB::optional_instant(&run, "execution_deadline_at")?, DB::optional_instant(&run, "run_deadline_at")?] - .into_iter() - .flatten() - .any(|deadline| deadline <= now()) - { - let response = Self::time_out_run(&mut tx, &run, &task, &body["workflow_task_attempt"]).await?; - tx.commit().await?; - self.wake.notify_waiters(); - return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response }); - } + let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?; // Workflow #930: the deadline close above is never metered. Before // the worker's first write, its local activities' exact events and // rows (other commands at one event) must fit the limits; then the diff --git a/rust/tests/execution/run_deadlines.rs b/rust/tests/execution/run_deadlines.rs index 344dedd7e..78dccecef 100644 --- a/rust/tests/execution/run_deadlines.rs +++ b/rust/tests/execution/run_deadlines.rs @@ -77,3 +77,98 @@ async fn workflow_task_before_the_deadline_completes_normally() { assert!(events_of(&history, "WorkflowTimedOut").is_empty()); assert_eq!(events_of(&history, "WorkflowCompleted").len(), 1); } + +// Server #794: a failure or history wait reported after the run's deadline +// closes the run as timed out, and the worker gets the same receipt a +// completion gets (PHP SDK 3.6.1 does not recognise a timeout refusal shaped +// as `waiting_for_history`). Published 2.11.8 +// acknowledged the wait and left the run waiting. +#[tokio::test] +async fn reports_after_a_deadline_time_out_the_run_with_the_completion_receipt() { + let waiting = json!({"type":"WorkflowTaskWaitingForHistory","message":"history is not ready yet"}); + let failure = json!({"type":"RuntimeException","message":"worker failed"}); + for kind in ["run", "execution"] { + let mut receipts = Vec::new(); + for report in ["waiting", "failure", "completion"] { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + let workflow_id = format!("deadline-{kind}-{report}"); + let (status, started) = request(&app, "POST", "/api/workflows", json!({"workflow_id":workflow_id,"workflow_type":"echo", + "task_queue":"test","input":envelope(Payload::Array(vec![])),format!("{kind}_timeout_seconds"):600})).await; + assert_eq!(status, StatusCode::CREATED, "{started}"); + let task = poll(&app, "worker", "workflow").await; + if kind == "run" { + database.execute("UPDATE workflow_runs SET run_deadline_at='2000-01-01 00:00:00'").await; + } else { + database.execute("UPDATE workflow_runs SET execution_deadline_at='2000-01-01 00:00:00'").await; + } + let task_id = task["task_id"].as_str().unwrap(); + let fail_path = format!("/api/worker/workflow-tasks/{task_id}/fail"); + let fail_body = |failure: &Value| json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"],"failure":failure}); + let history_before = run_history(&app, &json!(workflow_id), &started["run_id"]).await; + let mut stale = fail_body(&waiting); + stale["workflow_task_attempt"] = json!(task["workflow_task_attempt"].as_i64().unwrap() + 1); + let (status, refused) = request(&app, "POST", &fail_path, stale).await; + assert_eq!(status, StatusCode::CONFLICT, "{kind} {report}: {refused}"); + assert_eq!(run_history(&app, &json!(workflow_id), &started["run_id"]).await, history_before, "a stale attempt changes nothing"); + + let (status, mut receipt) = match report { + "waiting" => request(&app, "POST", &fail_path, fail_body(&waiting)).await, + "failure" => request(&app, "POST", &fail_path, fail_body(&failure)).await, + _ => finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Long(2))}])).await, + }; + assert_eq!(status, StatusCode::CONFLICT, "{kind} {report}: {receipt}"); + assert_eq!(receipt["task_id"], task["task_id"], "{receipt}"); + assert_eq!(receipt["workflow_task_attempt"], task["workflow_task_attempt"], "{receipt}"); + assert_eq!(receipt["run_id"], started["run_id"], "{receipt}"); + assert_eq!(receipt["outcome"], "completed", "{receipt}"); + assert_eq!(receipt["recorded"], false, "{receipt}"); + assert_eq!(receipt["run_status"], "failed", "{receipt}"); + assert_eq!(receipt["reason"], "run_timed_out", "{receipt}"); + assert_eq!(receipt["created_task_ids"], json!([]), "{receipt}"); + assert!(receipt["protocol_version"].is_string() && receipt["server_capabilities"].is_object(), "{receipt}"); + let (_, described) = request(&app, "GET", &format!("/api/workflows/{workflow_id}"), Value::Null).await; + assert_eq!((described["status"].as_str(), described["closed_reason"].as_str()), (Some("failed"), Some("timed_out")), "{described}"); + let history = run_history(&app, &json!(workflow_id), &started["run_id"]).await; + let timed_out = events_of(&history, "WorkflowTimedOut"); + assert_eq!(timed_out.len(), 1, "{history}"); + assert_eq!(timed_out[0]["payload"]["timeout_kind"], format!("{kind}_timeout")); + assert!(poll(&app, "worker", "workflow").await.is_null(), "no workflow task stays open"); + let (status, replay) = request(&app, "POST", &fail_path, fail_body(&waiting)).await; + assert_eq!(status, StatusCode::CONFLICT, "{replay}"); + let history = run_history(&app, &json!(workflow_id), &started["run_id"]).await; + assert_eq!(events_of(&history, "WorkflowTimedOut").len(), 1, "a replay adds no second timeout"); + for field in ["task_id", "run_id"] { + receipt[field] = Value::Null; + } + receipts.push(receipt); + } + assert_eq!(receipts[0], receipts[2], "{kind}: a history wait gets the completion's receipt"); + assert_eq!(receipts[1], receipts[2], "{kind}: a failure gets the completion's receipt"); + } +} + +// Before the deadline a history wait and a failure are acknowledged as usual. +#[tokio::test] +async fn reports_before_a_deadline_are_acknowledged() { + for (kind, failure, outcome) in [ + ("run", json!({"type":"WorkflowTaskWaitingForHistory","message":"history is not ready yet"}), "waiting_for_history"), + ("execution", json!({"type":"RuntimeException","message":"worker failed"}), "failed"), + ] { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + let workflow_id = format!("in-time-{kind}"); + let (status, started) = request(&app, "POST", "/api/workflows", json!({"workflow_id":workflow_id,"workflow_type":"echo", + "task_queue":"test","input":envelope(Payload::Array(vec![])),format!("{kind}_timeout_seconds"):600})).await; + assert_eq!(status, StatusCode::CREATED, "{started}"); + let task = poll(&app, "worker", "workflow").await; + let (status, answer) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/fail", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"],"failure":failure})).await; + assert_eq!(status, StatusCode::OK, "{kind}: {answer}"); + assert_eq!((answer["outcome"].as_str(), answer["recorded"].as_bool()), (Some(outcome), Some(true)), "{answer}"); + let history = run_history(&app, &json!(workflow_id), &started["run_id"]).await; + assert!(events_of(&history, "WorkflowTimedOut").is_empty(), "{history}"); + } +}