From 84e937470ead8b83faaed8d939b354eee24302a1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:01:47 -0600 Subject: [PATCH 1/2] Return the full not-recorded receipt for refused workflow-task reports (#994) A workflow-task completion, failure, waiting acknowledgement or cancellation-scope checkpoint refused task_not_leased or run_closed has written nothing. Its 409 now carries recorded:false, the task row's exact task_id and run_id, and the request's normalized workflow_task_attempt, beside the refusal's existing fields and task_status. This follows Security's contract in issue comment 6613395001. The guard and open-run refusals in fail_workflow_task and complete_or_checkpoint are mapped, as is Core's refused checkpoint for a run that closed after the guard. The identity fields are authoritative and overwrite anything the refusal carried. Attempt and owner mismatches, expired leases, stale registrations, heartbeats and history reads are unchanged and stay fatal to workers. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/store.rs | 53 +++- rust/src/runtime/workflow_task_guard.rs | 169 ++++++++++ rust/tests/execution.rs | 2 + rust/tests/execution/late_workflow_claims.rs | 20 +- rust/tests/execution/not_recorded_receipts.rs | 298 ++++++++++++++++++ 5 files changed, 529 insertions(+), 13 deletions(-) create mode 100644 rust/tests/execution/not_recorded_receipts.rs diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 054370bfd..816028756 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2699,9 +2699,13 @@ where if DB::string(&task, "task_type")? != "workflow" { return Err(refuse(StatusCode::NOT_FOUND, "task_not_found")); } - Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds).await?; + Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds) + .await + .map_err(|error| Self::not_recorded(&task, &body, error))?; let run_id = DB::string(&task, "workflow_run_id")?; - let run = Self::active_run(&mut tx, &run_id).await?; + let run = Self::active_run(&mut tx, &run_id) + .await + .map_err(|error| Self::not_recorded(&task, &body, error))?; // 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. @@ -3038,7 +3042,9 @@ where "outcome": "completed", "recorded": false, "reason": "already_completed", "created_task_ids": created}), ); } - Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds).await?; + Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds) + .await + .map_err(|error| Self::not_recorded(&task, &body, error))?; // PHP admits cancellation scopes before it guards worker-session // commands, and guards both before validating streams. Self::cancellation_scope_admission(&mut tx, &task, &body, commands, protocol).await?; @@ -3086,9 +3092,20 @@ where exceptions = super::local_activities::exceptions(&local, transition_codec.clone()).await?; } if let Some(checkpoint) = checkpoint { + // Server #994: Core's refused checkpoint for a run that closed + // after the guard is the same not-recorded receipt. + let identity = (DB::string(&task, "id")?, DB::string(&task, "workflow_run_id")?, DB::string(&task, "status")?); + let (receipt_task, receipt_run, receipt_status) = &identity; + let attempt = body["workflow_task_attempt"].clone(); return self .checkpoint_prefix(tx, task, &body, commands, checkpoint, transition_codec, &external, &local, &exceptions, limits, signal) - .await; + .await + .map(|answer| { + super::workflow_task_guard::not_recorded_answer(answer, receipt_task, &attempt, receipt_run, receipt_status) + }) + .map_err(|error| { + super::workflow_task_guard::not_recorded(error, receipt_task, &attempt, receipt_run, receipt_status) + }); } // `parseCommands`: at most one terminal command, anywhere in the // batch. It applies after every other command. @@ -3096,7 +3113,9 @@ where return Err(refuse(StatusCode::CONFLICT, "invalid_commands")); } let run_id = DB::string(&task, "workflow_run_id")?; - let run = Self::active_run(&mut tx, &run_id).await?; + let run = Self::active_run(&mut tx, &run_id) + .await + .map_err(|error| Self::not_recorded(&task, &body, error))?; // `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines // when one of its workflow tasks completes, and closes the run. let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?; @@ -4046,6 +4065,30 @@ where Err(refuse(StatusCode::CONFLICT, "conflicting_completion")) } + /// Server #994: a workflow task completion, failure or scope checkpoint + /// that the ownership guard or the open-run check refuses + /// `task_not_leased` or `run_closed` has written nothing (its transaction + /// is dropped), so its 409 carries the full not-recorded receipt: + /// `recorded: false` with the request's `workflow_task_attempt` and this + /// task row's own `task_id` and `run_id`, on top of the refusal's fields. + /// Workers settle on these fields alone and never infer them from the + /// reason. Every other refusal and error is returned as is. + fn not_recorded(task: &DB::Row, body: &Value, error: super::RuntimeError) -> super::RuntimeError { + let fields = (|| -> Result<(String, String, String)> { + Ok((DB::string(task, "id")?, DB::string(task, "workflow_run_id")?, DB::string(task, "status")?)) + })(); + match fields { + Ok((task_id, run_id, task_status)) => super::workflow_task_guard::not_recorded( + error, + &task_id, + &body["workflow_task_attempt"], + &run_id, + &task_status, + ), + Err(read) => read, + } + } + /// PHP's workflow task ownership guard, in PHP's order: a task that is /// no longer leased first (run_closed when its run has closed), then the /// lease's expiry, owner and attempt, and finally the owner's diff --git a/rust/src/runtime/workflow_task_guard.rs b/rust/src/runtime/workflow_task_guard.rs index b1b24b9f3..71fe3e2b2 100644 --- a/rust/src/runtime/workflow_task_guard.rs +++ b/rust/src/runtime/workflow_task_guard.rs @@ -24,3 +24,172 @@ pub(super) fn refusal( response, } } + +/// The refusals that settle a workflow task report as not recorded: the task +/// is no longer leased, or its run has closed. Every other refusal (lease +/// owner or attempt mismatch, expired lease, stale registration, conflicting +/// completion, invalid commands) stays fatal to a worker and gains nothing. +const NOT_RECORDED: [&str; 2] = ["task_not_leased", "run_closed"]; + +/// The full not-recorded receipt (Server #994, Security's contract in issue +/// comment 6613395001) on a workflow task completion, failure or scope +/// checkpoint refused with 409 `task_not_leased` or `run_closed` before +/// anything was written: `recorded: false` with the request's exact +/// `task_id` and `workflow_task_attempt`, the task row's own `run_id`, and +/// the task's status where the refusal has none. A guard refusal keeps every +/// field it already has; a bare refusal keeps its `reason` and `message`. No +/// payload or command data is added. Other errors pass through unchanged. +pub(super) fn not_recorded( + error: RuntimeError, + task_id: &str, + attempt: &Value, + run_id: &str, + task_status: &str, +) -> RuntimeError { + let mut response = match error { + RuntimeError::Protocol { status, response } + if status == StatusCode::CONFLICT + && response["reason"].as_str().is_some_and(|reason| NOT_RECORDED.contains(&reason)) => + { + response + } + RuntimeError::Refused { status, reason } if status == StatusCode::CONFLICT && NOT_RECORDED.contains(&reason) => { + json!({"reason":reason,"message":reason,"protocol_version":super::WORKER_PROTOCOL_VERSION, + "server_capabilities":super::store::capabilities()}) + } + other => return other, + }; + receipt(&mut response, task_id, attempt, run_id, task_status); + RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response, + } +} + +/// The same receipt on a refused answer the checkpoint returns as its body +/// (Core's refused checkpoint, `checkpointed: false`, rendered as 409): a +/// `task_not_leased` or `run_closed` checkpoint refusal gains the receipt; +/// any other answer is returned unchanged. +pub(super) fn not_recorded_answer( + mut answer: Value, + task_id: &str, + attempt: &Value, + run_id: &str, + task_status: &str, +) -> Value { + let refused = answer["checkpointed"] != json!(true) + && answer["reason"].as_str().is_some_and(|reason| NOT_RECORDED.contains(&reason)); + if refused { + receipt(&mut answer, task_id, attempt, run_id, task_status); + } + answer +} + +/// The receipt's identity is authoritative: the task row's id and run and +/// the request's normalized attempt replace whatever the refusal carried. +fn receipt(response: &mut Value, task_id: &str, attempt: &Value, run_id: &str, task_status: &str) { + if let Some(fields) = response.as_object_mut() { + let attempt = attempt.as_i64().map_or(Value::Null, Value::from); + fields.insert("task_id".to_owned(), json!(task_id)); + fields.insert("workflow_task_attempt".to_owned(), attempt); + fields.insert("recorded".to_owned(), json!(false)); + fields.insert("run_id".to_owned(), json!(run_id)); + fields.entry("task_status").or_insert_with(|| json!(task_status)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_guard_refusal_keeps_its_fields_and_gains_the_receipt() { + let guarded = refusal("task-1", &json!(2), "Workflow task is not currently leased.", "task_not_leased", + json!({"task_status":"completed"})); + let RuntimeError::Protocol { status, mut response } = not_recorded(guarded, "task-1", &json!(2), "run-1", "leased") else { + panic!("a protocol refusal"); + }; + assert_eq!(status, StatusCode::CONFLICT); + assert!(response.as_object_mut().unwrap().remove("server_capabilities").is_some()); + assert_eq!( + response, + json!({"task_id":"task-1","workflow_task_attempt":2,"error":"Workflow task is not currently leased.", + "reason":"task_not_leased","protocol_version":super::super::WORKER_PROTOCOL_VERSION, + "task_status":"completed","recorded":false,"run_id":"run-1"}) + ); + } + + #[test] + fn the_receipt_identity_is_authoritative_over_a_conflicting_body() { + let conflicting = refusal("other-task", &json!(9), "Workflow run is already closed.", "run_closed", + json!({"run_id":"other-run","recorded":true})); + let RuntimeError::Protocol { response, .. } = not_recorded(conflicting, "task-1", &json!(2), "run-1", "completed") else { + panic!("a protocol refusal"); + }; + assert_eq!( + (&response["task_id"], &response["workflow_task_attempt"], &response["run_id"], &response["recorded"]), + (&json!("task-1"), &json!(2), &json!("run-1"), &json!(false)) + ); + let answer = not_recorded_answer(json!({"checkpointed":false,"duplicate":false,"task_id":"other-task", + "workflow_task_attempt":9,"reason":"run_closed"}), "task-1", &json!(2), "run-1", "leased"); + assert_eq!( + answer, + json!({"checkpointed":false,"duplicate":false,"task_id":"task-1","workflow_task_attempt":2, + "reason":"run_closed","recorded":false,"run_id":"run-1","task_status":"leased"}) + ); + } + + #[test] + fn other_checkpoint_answers_are_unchanged() { + for answer in [ + json!({"checkpointed":false,"duplicate":false,"task_id":"task-1","reason":"workflow_claim_mismatch"}), + json!({"checkpointed":false,"reason":"active_claim_cancellation_not_supported"}), + json!({"checkpointed":true,"duplicate":true,"task_id":"task-1","reason":null}), + ] { + assert_eq!(not_recorded_answer(answer.clone(), "task-1", &json!(1), "run-1", "leased"), answer); + } + } + + #[test] + fn a_bare_run_closed_refusal_becomes_the_full_receipt() { + let RuntimeError::Protocol { status, mut response } = + not_recorded(super::super::refuse(StatusCode::CONFLICT, "run_closed"), "task-1", &json!(1), "run-1", "leased") + else { + panic!("a protocol refusal"); + }; + assert_eq!(status, StatusCode::CONFLICT); + assert!(response.as_object_mut().unwrap().remove("server_capabilities").is_some_and(|c| c.is_object())); + assert_eq!( + response, + json!({"reason":"run_closed","message":"run_closed","protocol_version":super::super::WORKER_PROTOCOL_VERSION, + "task_id":"task-1","workflow_task_attempt":1,"recorded":false,"run_id":"run-1","task_status":"leased"}) + ); + } + + #[test] + fn fatal_refusals_and_other_errors_pass_through_unchanged() { + for reason in ["lease_owner_mismatch", "workflow_task_attempt_mismatch", "lease_expired", "stale_worker_registration"] { + let guarded = refusal("task-1", &json!(1), "refused", reason, json!({})); + let RuntimeError::Protocol { response: expected, .. } = refusal("task-1", &json!(1), "refused", reason, json!({})) else { + unreachable!() + }; + let RuntimeError::Protocol { status, response } = not_recorded(guarded, "task-1", &json!(1), "run-1", "leased") else { + panic!("{reason}: unchanged"); + }; + assert_eq!((status, response), (StatusCode::CONFLICT, expected), "{reason}"); + } + for (status, reason) in [ + (StatusCode::CONFLICT, "conflicting_completion"), + (StatusCode::CONFLICT, "cancellation_deadline_expired"), + (StatusCode::UNPROCESSABLE_ENTITY, "invalid_request"), + (StatusCode::NOT_FOUND, "task_not_found"), + ] { + let RuntimeError::Refused { status: kept, reason: kept_reason } = + not_recorded(super::super::refuse(status, reason), "task-1", &json!(1), "run-1", "leased") + else { + panic!("{reason}: unchanged"); + }; + assert_eq!((kept, kept_reason), (status, reason)); + } + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 327e4a27c..c7b547bd4 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -43,6 +43,8 @@ mod namespaces; #[path = "execution/late_workflow_claims.rs"] mod late_workflow_claims; +#[path = "execution/not_recorded_receipts.rs"] +mod not_recorded_receipts; #[path = "execution/search_attributes.rs"] mod search_attributes; #[path = "execution/search_attribute_upserts.rs"] diff --git a/rust/tests/execution/late_workflow_claims.rs b/rust/tests/execution/late_workflow_claims.rs index 3e34d0d06..782b28c03 100644 --- a/rust/tests/execution/late_workflow_claims.rs +++ b/rust/tests/execution/late_workflow_claims.rs @@ -93,15 +93,19 @@ async fn a_replaced_workflow_claim_is_refused_with_the_published_ownership_field closed_at.ends_with('Z') && closed_at.len() == 27, "{action}: {refused}" ); - assert_eq!( - refused, - with( - json!({"error":"Workflow run is already closed.","reason":"run_closed","stop_reason":"run_completed", - "cancel_requested":false,"can_continue":false,"run_status":"completed","run_closed_reason":"completed", - "run_closed_at":closed_at,"task_status":"completed","lease_owner":"late-worker","lease_expires_at":null}) - ), - "{action}" + let mut expected = with( + json!({"error":"Workflow run is already closed.","reason":"run_closed","stop_reason":"run_completed", + "cancel_requested":false,"can_continue":false,"run_status":"completed","run_closed_reason":"completed", + "run_closed_at":closed_at,"task_status":"completed","lease_owner":"late-worker","lease_expires_at":null}), ); + // Server #994: a refused report also carries the full not-recorded + // receipt; a renewal keeps the guard body. + if action != "heartbeat" { + let fields = expected.as_object_mut().unwrap(); + fields.insert("recorded".into(), json!(false)); + fields.insert("run_id".into(), run["run_id"].clone()); + } + assert_eq!(refused, expected, "{action}"); } assert_eq!( request(&app, "GET", &history_path, Value::Null).await.1, diff --git a/rust/tests/execution/not_recorded_receipts.rs b/rust/tests/execution/not_recorded_receipts.rs new file mode 100644 index 000000000..b33ea7a31 --- /dev/null +++ b/rust/tests/execution/not_recorded_receipts.rs @@ -0,0 +1,298 @@ +use super::*; + +// Server #994, to Security's contract (issue comment 6613395001): a workflow +// task report (completion, failure or cancellation scope checkpoint) refused +// `task_not_leased` or `run_closed` has recorded nothing, and its 409 carries +// the full not-recorded receipt in the worker envelope: `recorded: false`, +// the request's exact `task_id` and `workflow_task_attempt`, and the task +// row's own `run_id`, beside the refusal's fields. Every other refusal stays +// fatal to a worker and gains none of it. + +const ROUTES: [&str; 3] = ["complete", "fail", "checkpoint"]; + +/// A late report on `route` for `task`, claimed by `owner` at `attempt`. +async fn report(app: &Router, task: &Value, route: &str, owner: &str, attempt: i64) -> (StatusCode, Value) { + let task_id = task["task_id"].as_str().unwrap(); + let claim = json!({"lease_owner":owner,"workflow_task_attempt":attempt}); + let with = |fields: Value| { + let mut body = claim.clone(); + body.as_object_mut().unwrap().extend(fields.as_object().unwrap().clone()); + body + }; + match route { + "complete" => { + let body = with(json!({"commands":[{"type":"complete_workflow","result":envelope(Payload::Long(7))}]})); + request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/complete"), body).await + } + "fail" => { + let body = with(json!({"failure":{"type":"RuntimeException","message":"late report"}})); + request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/fail"), body).await + } + "waiting" => { + let body = with(json!({"failure":{"type":"WorkflowTaskWaitingForHistory","message":"history is not ready yet"}})); + request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/fail"), body).await + } + _ => { + let body = with(json!({"checkpoint_id":"late","start_sequence":1, + "commands":[{"type":"start_timer","delay_seconds":60}]})); + let path = format!("/api/worker/workflow-tasks/{task_id}/cancellation-scopes/checkpoint"); + request_protocol(app, "POST", &path, body, "1.20").await + } + } +} + +/// The refusal's public worker body without its capability manifest, which +/// must be present. +fn public_body(refused: &(StatusCode, Value), label: &str) -> Value { + let (status, body) = refused; + assert_eq!(*status, StatusCode::CONFLICT, "{label}: {body}"); + let mut body = body.clone(); + let fields = body.as_object_mut().unwrap(); + assert!(fields.remove("server_capabilities").is_some_and(|manifest| manifest.is_object()), "{label}: {body}"); + body +} + +/// The exact receipt a not-recorded refusal must carry. +fn receipt(task: &Value, attempt: i64) -> Value { + json!({"task_id":task["task_id"],"workflow_task_attempt":attempt,"recorded":false,"run_id":task["run_id"]}) +} + +fn merged(base: Value, extra: Value) -> Value { + let mut base = base; + base.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone()); + base +} + +async fn claimed(app: &Router, workflow: &str) -> (Value, Value) { + register(app, "nr-worker", json!(["echo"]), json!([])).await; + let started = start(app, workflow).await; + let task = poll(app, "nr-worker", "workflow").await; + assert_eq!(task["run_id"], started["run_id"], "{task}"); + assert!(task["run_id"].as_str().is_some_and(|run| !run.is_empty())); + (started, task) +} + +/// Durable state a late report must leave unchanged: every table's row +/// count, the run's history and the stored completion receipts. +async fn durable(database: &TestDatabase, app: &Router, started: &Value) -> Value { + json!({"tables":table_counts(database).await, + "history":run_history(app, &started["workflow_id"], &started["run_id"]).await, + "receipts":text_rows(database, "SELECT task_id, receipt FROM dw_task_completions ORDER BY task_id", + "SELECT task_id, receipt::text FROM dw_task_completions ORDER BY task_id", + "SELECT task_id, CAST(receipt AS CHAR) FROM dw_task_completions ORDER BY task_id").await}) +} + +/// `run_closed` for every terminal status the guard recognizes, on every +/// report route: the guard's closed-run body plus the receipt, nothing +/// recorded. +#[tokio::test] +async fn a_report_on_every_closed_run_status_is_run_closed_with_the_full_receipt() { + for close in ["completed", "failed", "timed_out", "cancelled", "terminated"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let workflow = format!("nr-closed-{close}"); + let (started, task) = claimed(&app, &workflow).await; + let run_id = started["run_id"].as_str().unwrap(); + // How the run closes while the worker's report is late. + let (run_status, stop_reason, closed_reason, task_status, cancel_requested) = match close { + "completed" | "failed" => { + let command = if close == "completed" { + json!({"type":"complete_workflow","result":envelope(Payload::Long(1))}) + } else { + json!({"type":"fail_workflow","message":"boom"}) + }; + let done = finish_task(&app, &task, json!([command])).await; + assert_eq!(done.0, StatusCode::OK, "{close}: {}", done.1); + // A terminal fail_workflow leaves its task failed. + let (stop, task_status) = if close == "completed" { ("run_completed", "completed") } else { ("run_failed", "failed") }; + (close, stop, close, task_status, false) + } + "timed_out" => { + database.execute("UPDATE workflow_runs SET run_deadline_at='2000-01-01 00:00:00'").await; + let timed_out = report(&app, &task, "complete", "nr-worker", 1).await; + assert_eq!( + (timed_out.0, &timed_out.1["reason"], &timed_out.1["recorded"]), + (StatusCode::CONFLICT, &json!("run_timed_out"), &json!(false)), + "{}", + timed_out.1 + ); + ("failed", "run_failed", "timed_out", "completed", false) + } + _ => { + let path = format!("/api/workflows/{workflow}/runs/{run_id}/{}", if close == "cancelled" { "cancel" } else { "terminate" }); + let closed = request(&app, "POST", &path, json!({"reason":"operator"})).await; + assert_eq!(closed.0, StatusCode::OK, "{close}: {}", closed.1); + let stop = if close == "cancelled" { "run_cancelled" } else { "run_terminated" }; + (close, stop, close, "cancelled", true) + } + }; + let before = durable(&database, &app, &started).await; + for route in ROUTES { + let label = format!("{close} {route}"); + let mut body = public_body(&report(&app, &task, route, "nr-worker", 1).await, &label); + let closed_at = body.as_object_mut().unwrap().remove("run_closed_at").unwrap_or_default(); + assert!(closed_at.as_str().is_some_and(|at| at.ends_with('Z') && at.len() == 27), "{label}: {closed_at}"); + assert_eq!( + body, + merged( + receipt(&task, 1), + json!({"error":"Workflow run is already closed.","reason":"run_closed","protocol_version":"1.20", + "stop_reason":stop_reason,"cancel_requested":cancel_requested,"can_continue":false, + "run_status":run_status,"run_closed_reason":closed_reason,"task_status":task_status, + "lease_owner":"nr-worker","lease_expires_at":null}) + ), + "{label}" + ); + } + assert_eq!(durable(&database, &app, &started).await, before, "{close}: the late reports recorded nothing"); + runtime.close().await; + database.remove().await; + } +} + +/// `task_not_leased` for a task completed on a run that is still open, and +/// for the waiting acknowledgement's duplicate path, whose task holds no +/// completion receipt. +#[tokio::test] +async fn a_report_for_a_finished_task_on_an_open_run_is_task_not_leased_with_the_full_receipt() { + for finish in ["completed", "waiting"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (started, task) = claimed(&app, &format!("nr-not-leased-{finish}")).await; + if finish == "completed" { + let done = finish_task(&app, &task, json!([{"type":"start_timer","delay_seconds":3600}])).await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + } else { + let acknowledged = report(&app, &task, "waiting", "nr-worker", 1).await; + assert_eq!( + (acknowledged.0, &acknowledged.1["outcome"]), + (StatusCode::OK, &json!("waiting_for_history")), + "{}", + acknowledged.1 + ); + } + let before = durable(&database, &app, &started).await; + for route in ["complete", "fail", "waiting", "checkpoint"] { + let label = format!("{finish} {route}"); + assert_eq!( + public_body(&report(&app, &task, route, "nr-worker", 1).await, &label), + merged( + receipt(&task, 1), + json!({"error":"Workflow task is not currently leased.","reason":"task_not_leased", + "protocol_version":"1.20","task_status":"completed"}) + ), + "{label}" + ); + } + assert_eq!(durable(&database, &app, &started).await, before, "{finish}: the late reports recorded nothing"); + runtime.close().await; + database.remove().await; + } +} + +/// A run closed by another path while its task kept the lease passes the +/// ownership guard and is refused by the open-run check: the bare refusal +/// becomes the same receipt in the worker envelope. +#[tokio::test] +async fn a_report_on_a_closed_run_whose_task_is_still_leased_carries_the_full_receipt() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (started, task) = claimed(&app, "nr-leased-closed").await; + database.execute("UPDATE workflow_runs SET status='failed'").await; + let before = durable(&database, &app, &started).await; + for route in ["complete", "fail"] { + assert_eq!( + public_body(&report(&app, &task, route, "nr-worker", 1).await, route), + merged( + receipt(&task, 1), + json!({"reason":"run_closed","message":"run_closed","protocol_version":"1.20","task_status":"leased"}) + ), + "{route}" + ); + } + assert_eq!(durable(&database, &app, &started).await, before); + runtime.close().await; + database.remove().await; +} + +/// The refusals that are not settlement authority keep their published +/// bodies: no `recorded`, no `run_id`. +#[tokio::test] +async fn fatal_report_refusals_gain_no_receipt() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (started, replaced) = claimed(&app, "nr-fatal").await; + // Registering the same worker again releases its lease; the task is + // claimed again as attempt 2. + register(&app, "nr-worker", json!(["echo"]), json!([])).await; + let current = poll(&app, "nr-worker", "workflow").await; + assert_eq!((¤t["task_id"], ¤t["workflow_task_attempt"]), (&replaced["task_id"], &json!(2))); + let before = durable(&database, &app, &started).await; + let fence = |attempt: i64, fields: Value| { + merged(json!({"task_id":replaced["task_id"],"workflow_task_attempt":attempt,"protocol_version":"1.20"}), fields) + }; + for route in ROUTES { + assert_eq!( + public_body(&report(&app, &replaced, route, "nr-worker", 1).await, route), + fence(1, json!({"error":"Workflow task lease attempt does not match the current claim.", + "reason":"workflow_task_attempt_mismatch","current_attempt":2})), + "{route}: attempt mismatch" + ); + assert_eq!( + public_body(&report(&app, &replaced, route, "nr-other-worker", 2).await, route), + fence(2, json!({"error":"Workflow task lease is owned by another worker.", + "reason":"lease_owner_mismatch","lease_owner":"nr-worker"})), + "{route}: owner mismatch" + ); + } + database.set_time(false, "2000-01-01T00:00:00Z").await; + for route in ROUTES { + let body = public_body(&report(&app, &replaced, route, "nr-worker", 2).await, route); + assert_eq!(body["reason"], "lease_expired", "{route}: {body}"); + assert!(body.get("recorded").is_none() && body.get("run_id").is_none(), "{route}: {body}"); + } + assert_eq!(durable(&database, &app, &started).await, before); + runtime.close().await; + database.remove().await; +} + +/// The shared checkpoint wrapper: a cooperative claim still leased whose +/// run closes after the ownership guard reaches Core's checkpoint, whose +/// refused `run_closed` answer carries the same receipt. +#[tokio::test] +async fn a_checkpoint_whose_run_closed_after_the_guard_carries_the_full_receipt() { + for status in ["completed", "failed", "cancelled", "terminated"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let registered = request_protocol(&app, "POST", "/api/worker/register", + json!({"worker_id":"nr-worker","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":["cooperative_cancellation"]}), "1.20").await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let started = start(&app, &format!("nr-checkpoint-{status}")).await; + let claimed = request_protocol(&app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"nr-worker","task_queue":"test","timeout_seconds":2}), "1.20").await; + let task = claimed.1["task"].clone(); + assert_eq!(task["run_id"], started["run_id"], "{}", claimed.1); + let sql: &'static str = Box::leak(format!("UPDATE workflow_runs SET status='{status}'").into_boxed_str()); + database.execute(sql).await; + let before = durable(&database, &app, &started).await; + let label = format!("{status} checkpoint"); + assert_eq!( + public_body(&report(&app, &task, "checkpoint", "nr-worker", 1).await, &label), + merged( + receipt(&task, 1), + json!({"checkpointed":false,"duplicate":false,"reason":"run_closed","task_status":"leased", + "history_refresh_page_token":"MA==","protocol_version":"1.20"}) + ), + "{label}" + ); + assert_eq!(durable(&database, &app, &started).await, before, "{label}: nothing recorded"); + runtime.close().await; + database.remove().await; + } +} From b4654d1a33a7adb1a2288e8cc3d5a6af0a48c7a6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:52:51 -0600 Subject: [PATCH 2/2] Late completion conformance: expect the not-recorded receipt from native only (#994) Published PHP 2.11.8's ownership guard refuses a late completion or failure, and a closed-run report, without recorded or run_id. It reports both only from its rarer post-guard engine outcome, which a deterministic late request never reaches. Native now adds the not-recorded receipt to those refusals (server#994). The probe records the implementation from cluster info (a failed read throws). The contract requires PHP's published bodies exactly when it is absent, and for native the same bodies plus exactly recorded false and the task's run_id. Renewal and attempt-mismatch refusals are unchanged for both. Any other implementation is refused. Unit controls: PHP with native fields, native without them (3.0.4's bodies), a wrong run, a claimed record, receipt fields on renewal or attempt mismatch, and unknown or unnamed implementations all fail. Against main's contract the native case and the 3.0.4 control fail; without the implementation check, both unknown-implementation controls fail. Co-Authored-By: Claude Opus 5.5 --- .../late-workflow-completion-contract.mjs | 21 +++++++-- .../late-workflow-completion-probe.php | 6 ++- .../late-workflow-completion.json | 2 +- ...rityLateWorkflowCompletionContractTest.mjs | 44 ++++++++++++++++++- 4 files changed, 66 insertions(+), 7 deletions(-) diff --git a/scripts/conformance/server-parity/late-workflow-completion-contract.mjs b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs index e332d56ee..c4fc62bc7 100644 --- a/scripts/conformance/server-parity/late-workflow-completion-contract.mjs +++ b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs @@ -40,10 +40,23 @@ export function checkLateWorkflowCompletion(fixture, observation, workflowId) { equal(state.heartbeat_while_leased.status, 409, 'late lease renewal refused while the new claim holds the lease'); equal(body(state.heartbeat_while_leased), body(state.while_leased), 'renewal refusal matches the completion refusal'); - for (const key of ['after_completion', 'heartbeat_after_completion', 'failure_after_completion']) { + // Published PHP's ownership guard (lines 4737-4775 at tag 2.11.8) sends + // neither `recorded` nor `run_id` here; it reports them only from its rarer + // post-guard engine outcome. Native adds the not-recorded receipt to every + // refused report (server#994), naming the run from the task row. Lease + // renewals record nothing either way and keep the published body. + const {implementation} = state; + assert.ok(implementation === null || implementation === 'rust', `published PHP or native, not ${implementation}`); + if (implementation === 'rust') { + assert.ok(typeof state.peer_run_id === 'string' && state.peer_run_id !== '', 'authoritative run id'); + } + const receipt = implementation === 'rust' ? {recorded: false, run_id: state.peer_run_id} : {}; + const notLeased = {...fence, error: 'Workflow task is not currently leased.', reason: 'task_not_leased', + task_status: 'completed'}; + for (const [key, expected] of [['after_completion', {...notLeased, ...receipt}], + ['heartbeat_after_completion', notLeased], ['failure_after_completion', {...notLeased, ...receipt}]]) { equal(state[key].status, 409, `${key} refused`); - equal(body(state[key]), {...fence, error: 'Workflow task is not currently leased.', reason: 'task_not_leased', - task_status: 'completed'}, `${key}: the completed task is not leased`); + equal(body(state[key]), expected, `${key}: the completed task is not leased (${implementation ?? 'php'})`); } equal(state.history_while_open.map(event => event.event_type), definition.expected_events.slice(0, 3), 'late requests recorded nothing while the run was open'); @@ -56,7 +69,7 @@ export function checkLateWorkflowCompletion(fixture, observation, workflowId) { equal(closedFields, {task_id: lastReplaced.task_id, workflow_task_attempt: 1, error: 'Workflow run is already closed.', reason: 'run_closed', stop_reason: 'run_completed', cancel_requested: false, can_continue: false, run_status: 'completed', run_closed_reason: 'completed', task_status: 'completed', lease_owner: worker, - lease_expires_at: null}, 'run closed refusal: the completed task keeps its owner and has no lease expiry'); + lease_expires_at: null, ...receipt}, 'run closed refusal: the completed task keeps its owner and has no lease expiry'); equal(state.history.map(event => event.event_type), definition.expected_events, 'one completion recorded'); return projection; } diff --git a/scripts/conformance/server-parity/late-workflow-completion-probe.php b/scripts/conformance/server-parity/late-workflow-completion-probe.php index de65b956d..95bcf8f6d 100644 --- a/scripts/conformance/server-parity/late-workflow-completion-probe.php +++ b/scripts/conformance/server-parity/late-workflow-completion-probe.php @@ -41,7 +41,11 @@ function finishHttpLateWorkflowCompletion(Client $client, array $fixture, string $complete = [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->envelope($fixture['input'])]]; $late = static fn (array $task, array $commands): array => $receipt(static fn (): array => $client->completeWorkflowTask( $task['task_id'], $worker, $task['workflow_task_attempt'], $commands)); - $state = ['applicable' => true, 'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId]; + // Native adds the not-recorded receipt (server#994); published PHP omits + // `implementation` from cluster info. A failed read throws, so the + // recording never guesses which bodies to expect. + $state = ['applicable' => true, 'implementation' => $client->clusterInfo()->raw['implementation'] ?? null, + 'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId]; $register(); $handle = $client->startWorkflow($definition['workflow_type'], $peerId, $queue, [$fixture['input']]); $state['peer_run_id'] = $handle->selectedRunId; diff --git a/tests/Fixtures/ServerParity/late-workflow-completion.json b/tests/Fixtures/ServerParity/late-workflow-completion.json index 42006e41d..28d51b1e2 100644 --- a/tests/Fixtures/ServerParity/late-workflow-completion.json +++ b/tests/Fixtures/ServerParity/late-workflow-completion.json @@ -43,5 +43,5 @@ "WorkflowCompleted" ] }, - "contract": "A worker process holding a workflow task claim is replaced by a new registration of the same worker_id, which releases the lease; the task is claimed again as attempt 2. Requests from the replaced attempt 1 are refused with the published ownership guard bodies a worker reads to recognize a lost claim. While attempt 2 holds the lease, completion and lease renewal are refused with 409 workflow_task_attempt_mismatch and the current attempt. After attempt 2 completes the task while the run is open, completion, renewal and failure are refused with 409 task_not_leased and task_status completed. After the run closes, a late completion is refused with 409 run_closed, can_continue false, the run status and closure, and the completed task, which keeps its lease owner and has no lease expiry. No late request records history; each task records one completion. Embedded applications have no HTTP worker registration." + "contract": "A worker process holding a workflow task claim is replaced by a new registration of the same worker_id, which releases the lease; the task is claimed again as attempt 2. Requests from the replaced attempt 1 are refused with the published ownership guard bodies a worker reads to recognize a lost claim. While attempt 2 holds the lease, completion and lease renewal are refused with 409 workflow_task_attempt_mismatch and the current attempt. After attempt 2 completes the task while the run is open, completion, renewal and failure are refused with 409 task_not_leased and task_status completed. After the run closes, a late completion is refused with 409 run_closed, can_continue false, the run status and closure, and the completed task, which keeps its lease owner and has no lease expiry. No late request records history; each task records one completion. Published PHP's ownership guard sends these refusals without recorded or run_id (it reports both only from its rarer post-guard engine outcome, which a deterministic late request does not reach); native adds the not-recorded receipt, exactly recorded false and the task's run_id, to the refused completion, failure and closed-run reports, while renewal refusals keep the published body. The recording names the implementation from cluster info, and any other implementation is refused. Embedded applications have no HTTP worker registration." } diff --git a/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs index 848d689f0..3119e3d1c 100644 --- a/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs +++ b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs @@ -13,7 +13,7 @@ function model() { reason: 'workflow_task_attempt_mismatch', current_attempt: 2}); const notLeased = refusal({error: 'Workflow task is not currently leased.', reason: 'task_not_leased', task_status: 'completed'}); const events = fixture.late_workflow_completion.expected_events.map(event_type => ({event_type})); - return structuredClone({mode: 'http', late_workflow_completion: {applicable: true, worker_id: worker, + return structuredClone({mode: 'http', late_workflow_completion: {applicable: true, implementation: null, worker_id: worker, task_queue: 'test-late', peer_workflow_id: 'test-late-peer', peer_run_id: 'peer-run', first_claims: claims('task-1'), while_leased: mismatch, heartbeat_while_leased: mismatch, scheduled: {recorded: true}, after_completion: notLeased, heartbeat_after_completion: notLeased, @@ -24,6 +24,48 @@ function model() { run_closed_at: '2026-10-20T17:33:21.305246Z', task_status: 'completed', lease_owner: worker, lease_expires_at: null}}, history: events}}); } +// Native's bodies: the published ones plus the not-recorded receipt on each +// refused report (server#994). Renewal and attempt-mismatch bodies are unchanged. +function nativeModel() { + const observation = model(); + const state = observation.late_workflow_completion; + state.implementation = 'rust'; + for (const key of ['after_completion', 'failure_after_completion', 'after_close']) { + Object.assign(state[key].response, {recorded: false, run_id: state.peer_run_id}); + } + state.heartbeat_after_completion = structuredClone(model().late_workflow_completion.heartbeat_after_completion); + return observation; +} +test('native lost-claim refusals with the not-recorded receipt pass', () => + assert.equal(checkLateWorkflowCompletion(fixture, nativeModel(), 'test').scope, 'late_workflow_completion')); +for (const [label, build, mutate] of [ + ['published PHP refusals carrying native receipt fields', model, + s => Object.assign(s.after_completion.response, {recorded: false, run_id: s.peer_run_id})], + ['published PHP closed-run refusal carrying recorded', model, s => s.after_close.response.recorded = false], + ['native refusals without the receipt (3.0.4 bodies)', model, s => s.implementation = 'rust'], + ['native completion refusal without run_id', nativeModel, s => delete s.after_completion.response.run_id], + ['native failure refusal without recorded', nativeModel, s => delete s.failure_after_completion.response.recorded], + ['native closed-run refusal without the receipt', nativeModel, + s => { delete s.after_close.response.recorded; delete s.after_close.response.run_id; }], + ['native receipt that claims a record', nativeModel, s => s.after_completion.response.recorded = true], + ['native receipt naming another run', nativeModel, s => s.after_close.response.run_id = 'other-run'], + ['native renewal refusal carrying the receipt', nativeModel, + s => Object.assign(s.heartbeat_after_completion.response, {recorded: false, run_id: s.peer_run_id})], + ['native attempt mismatch carrying the receipt', nativeModel, + s => { Object.assign(s.while_leased.response, {recorded: false, run_id: s.peer_run_id}); + Object.assign(s.heartbeat_while_leased.response, {recorded: false, run_id: s.peer_run_id}); }], + ['native without an authoritative run id', nativeModel, s => s.peer_run_id = ''], + // Published bodies under an unrecognized name: only the implementation check refuses them. + ['an unknown implementation with published bodies', model, s => s.implementation = 'php'], + ['an unknown implementation with native bodies', nativeModel, s => s.implementation = 'native'], + ['an implementation the recording did not name', model, s => delete s.implementation], +]) { + test(`rejects ${label}`, () => { + const observation = build(); + mutate(observation.late_workflow_completion); + assert.throws(() => checkLateWorkflowCompletion(fixture, observation, 'test')); + }); +} test('published lost-claim refusals pass', () => assert.equal(checkLateWorkflowCompletion(fixture, model(), 'test').scope, 'late_workflow_completion')); test('embedded explicitly declares the HTTP worker lifecycle inapplicable', () => checkLateWorkflowCompletion(fixture,