From 9e361afb3c418dcc1ba4443f69b442bf24251c37 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 21:30:43 +0000 Subject: [PATCH] Renew activity leases like Server 2.11 POST /api/worker/activity-tasks/{task}/renew-lease keeps a live attempt's lease for its owner without recording progress, moving a business deadline or adding history, as published Server 2.11.1 does over Workflow 2.10's ActivityLeaseRenewal: PHP's ownership guard and bodies, Laravel's validation, the attempt status with lease_renewed and server_time, and refusals for closed, cancelled, terminated or expired authority. Native advertises activity_lease_renewal, so current SDKs use it. Closing an attempt now clears its lease and its task's, as PHP's closeAttempt does; completed attempts and their tasks kept a stale lease. The activity-lease-renewal fixture records each case against the published reference. --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 4 +- rust/src/runtime/activity_failures.rs | 2 +- rust/src/runtime/activity_lease_renewal.rs | 283 ++++++++++++++++++ rust/src/runtime/execution.rs | 1 + rust/src/runtime/http.rs | 14 +- rust/src/runtime/mod.rs | 22 ++ rust/src/runtime/store.rs | 9 +- rust/tests/execution.rs | 3 + .../tests/execution/activity_lease_renewal.rs | 173 +++++++++++ .../activity-lease-renewal-contract.mjs | 113 +++++++ .../activity-lease-renewal-probe.php | 111 +++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../ServerParity/activity-lease-renewal.json | 37 +++ ...ParityActivityLeaseRenewalContractTest.mjs | 110 +++++++ 16 files changed, 887 insertions(+), 8 deletions(-) create mode 100644 rust/src/runtime/activity_lease_renewal.rs create mode 100644 rust/tests/execution/activity_lease_renewal.rs create mode 100644 scripts/conformance/server-parity/activity-lease-renewal-contract.mjs create mode 100644 scripts/conformance/server-parity/activity-lease-renewal-probe.php create mode 100644 tests/Fixtures/ServerParity/activity-lease-renewal.json create mode 100644 tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index d962afadb..bf728d4b1 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -34,6 +34,7 @@ on: - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWorkerReplacementContractTest.mjs - tests/Unit/ServerParityOperatorMetricsContractTest.mjs + - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -81,6 +82,7 @@ on: - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWorkerReplacementContractTest.mjs - tests/Unit/ServerParityOperatorMetricsContractTest.mjs + - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1498,7 +1500,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 25652b850..db70a80bf 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -112,7 +112,9 @@ These published PHP routes on this tree have no native handler yet, by family. - Local activities (prepare, checkpoint, recover, heartbeat, control, outcome and cancellation acknowledgement). - Cancellation scopes. - - Activity lease renewal and cancellation acknowledgement. + - Activity cancellation acknowledgement. (Lease renewal, from Server 2.11, + is served: `POST /api/worker/activity-tasks/{task}/renew-lease`, with the + `activity_lease_renewal` capability and the `activity-lease-renewal` fixture.) - **Run operations.** - Archive (instance and run). - Redrive. diff --git a/rust/src/runtime/activity_failures.rs b/rust/src/runtime/activity_failures.rs index 9ff800835..1dca08ace 100644 --- a/rust/src/runtime/activity_failures.rs +++ b/rust/src/runtime/activity_failures.rs @@ -464,7 +464,7 @@ where "invalid_activity_retry_policy", ) })?; - Self::query("UPDATE activity_attempts SET status='failed',closed_at=$1 WHERE id=$2") + Self::query("UPDATE activity_attempts SET status='failed',lease_expires_at=NULL,closed_at=$1 WHERE id=$2") .bind(DB::bind_time(closed_at)) .bind(attempt_id) .execute(&mut *tx) diff --git a/rust/src/runtime/activity_lease_renewal.rs b/rust/src/runtime/activity_lease_renewal.rs new file mode 100644 index 000000000..83890c97d --- /dev/null +++ b/rust/src/runtime/activity_lease_renewal.rs @@ -0,0 +1,283 @@ +//! Runtime activity lease renewal, as published Server 2.11 serves it +//! (`ActivityTaskController::renewLease` over Workflow 2.10's +//! `ActivityLeaseRenewal`). Renewal keeps a live attempt's lease for its +//! original owner. It never records progress, moves a business deadline or +//! adds history, and an expired lease cannot be revived. +use super::laravel_rules::{self, Rule}; +use super::transaction::Transaction; +use super::{ + Result, RuntimeError, + store::{capabilities, now}, +}; +use axum::http::StatusCode; +use chrono::{DateTime, SecondsFormat, Utc}; +use serde_json::{Value, json}; + +/// Published PHP's validation rules for the request, in their order. +const RULES: &laravel_rules::Rules = &[ + ("activity_attempt_id", &[Rule::Required, Rule::String]), + ("lease_owner", &[Rule::Required, Rule::String]), +]; + +/// A worker-protocol response with PHP's protocol fields. +fn worker(status: StatusCode, mut response: Value) -> RuntimeError { + response["protocol_version"] = json!(super::WORKER_PROTOCOL_VERSION); + response["server_capabilities"] = capabilities(); + RuntimeError::Protocol { status, response } +} + +/// PHP's `outcomeStatus`. +fn outcome_status(reason: Option<&str>) -> StatusCode { + match reason { + None => StatusCode::OK, + Some("attempt_not_found") => StatusCode::NOT_FOUND, + Some(_) => StatusCode::CONFLICT, + } +} + +fn micros(at: DateTime) -> String { + at.to_rfc3339_opts(SecondsFormat::Micros, true) +} + +/// The rows an attempt's authority is read from. +struct Rows { + attempt: R, + activity: Option, + run: Option, + task: Option, +} + +store_methods! { + pub(crate) async fn renew_activity_lease(&self, task_id: &str, body: Value) -> Result { + let body = laravel_rules::normalize(&body); + if let Err(RuntimeError::Protocol { status, response }) = laravel_rules::validate(&body, RULES) { + return Err(worker(status, response)); + } + let attempt_id = body["activity_attempt_id"].as_str().unwrap_or_default(); + let owner = body["lease_owner"].as_str().unwrap_or_default(); + let refusal = |status, error: &str, reason: &str| { + worker(status, json!({"task_id":task_id,"activity_attempt_id":attempt_id,"error":error,"reason":reason})) + }; + let mut tx = self.begin().await?; + // PHP's ownership guard, in its order: the task in this namespace, + // then the attempt by its identity, its task and its owner. + let task = Self::query("SELECT id FROM workflow_tasks WHERE id=$1 AND namespace=$2") + .bind(task_id) + .bind(self.namespace.as_ref()) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| refusal(StatusCode::NOT_FOUND, "Activity task not found.", "task_not_found"))?; + let Some(rows) = Self::attempt_rows(&mut tx, attempt_id).await? else { + return Err(refusal(StatusCode::NOT_FOUND, "Activity attempt not found.", "attempt_not_found")); + }; + let observed = Self::attempt_status(attempt_id, &rows, None)?; + if observed["workflow_task_id"].as_str() != Some(DB::string(&task, "id")?.as_str()) { + return Err(refusal(StatusCode::CONFLICT, "Activity attempt is not leased for this task.", "task_mismatch")); + } + if observed["lease_owner"].as_str() != Some(owner) { + let mut error = json!({"task_id":task_id,"activity_attempt_id":attempt_id, + "error":"Activity attempt lease is owned by another worker.","reason":"lease_owner_mismatch"}); + error["lease_owner"] = observed["lease_owner"].clone(); + return Err(worker(StatusCode::CONFLICT, error)); + } + if observed["can_continue"] != true { + let mut response = observed; + response["lease_renewed"] = json!(false); + return Err(worker(outcome_status(response["reason"].as_str()), response)); + } + // The renewal itself requires the owner of both the attempt and its + // task, as PHP's observation with an owner does. + let leased_by_owner = DB::optional_string(&rows.attempt, "lease_owner")?.as_deref() == Some(owner) + && match &rows.task { + Some(task) => DB::optional_string(task, "lease_owner")?.as_deref() == Some(owner), + None => false, + }; + let at = now(); + let mut status = if leased_by_owner { + let task = rows.task.as_ref().expect("task checked above"); + let mut expires = at + chrono::Duration::seconds(self.activity_task_lease_seconds); + if let Some(current) = DB::optional_instant(task, "lease_expires_at")?.filter(|current| *current > expires) { + expires = current; + } + Self::query("UPDATE activity_attempts SET lease_expires_at=$1 WHERE id=$2") + .bind(DB::bind_time(expires)) + .bind(attempt_id) + .execute(&mut *tx) + .await?; + Self::query("UPDATE workflow_tasks SET lease_expires_at=$1 WHERE id=$2") + .bind(DB::bind_time(expires)) + .bind(DB::string(task, "id")?) + .execute(&mut *tx) + .await?; + let renewed = Self::attempt_rows(&mut tx, attempt_id).await?.expect("attempt read above"); + Self::attempt_status(attempt_id, &renewed, None)? + } else { + Self::attempt_status(attempt_id, &rows, Some("lease_owner_mismatch"))? + }; + tx.commit().await?; + status["lease_renewed"] = status["can_continue"].clone(); + status["server_time"] = json!(micros(at)); + status["task_id"] = json!(task_id); + status["worker_session"] = Value::Null; + let code = outcome_status(status["reason"].as_str()); + if code != StatusCode::OK { + return Err(worker(code, status)); + } + status["protocol_version"] = json!(super::WORKER_PROTOCOL_VERSION); + status["server_capabilities"] = capabilities(); + Ok(status) + } + + async fn attempt_rows(tx: &mut Transaction<'_, DB>, attempt_id: &str) -> Result>> { + let Some(attempt) = Self::query("SELECT * FROM activity_attempts WHERE id=$1") + .bind(attempt_id) + .fetch_optional(&mut **tx) + .await? + else { + return Ok(None); + }; + let activity = Self::query("SELECT * FROM activity_executions WHERE id=$1") + .bind(DB::string(&attempt, "activity_execution_id")?) + .fetch_optional(&mut **tx) + .await?; + let run = match &activity { + Some(activity) => Self::query("SELECT * FROM workflow_runs WHERE id=$1") + .bind(DB::string(activity, "workflow_run_id")?) + .fetch_optional(&mut **tx) + .await?, + None => None, + }; + let task = match DB::optional_string(&attempt, "workflow_task_id")?.filter(|id| !id.is_empty()) { + Some(id) => Self::query("SELECT * FROM workflow_tasks WHERE id=$1") + .bind(id) + .fetch_optional(&mut **tx) + .await?, + None => None, + }; + Ok(Some(Rows { attempt, activity, run, task })) + } + + /// Workflow's `attemptStatus` over its continuation state. `refused` + /// replaces the state with a refusal that cannot continue. + fn attempt_status(attempt_id: &str, rows: &Rows, refused: Option<&'static str>) -> Result { + let (can_continue, cancel_requested, reason) = match refused { + Some(reason) => (false, false, Some(reason)), + None => Self::continuation(rows, now())?, + }; + let text = |row: Option<&DB::Row>, column: &str| -> Result> { + Ok(match row { + Some(row) => DB::optional_string(row, column)?.filter(|value| !value.is_empty()), + None => None, + }) + }; + let instant = |row: Option<&DB::Row>, column: &str| -> Result>> { + Ok(match row { + Some(row) => DB::optional_instant(row, column)?, + None => None, + }) + }; + let attempt = Some(&rows.attempt); + let (activity, run, task) = (rows.activity.as_ref(), rows.run.as_ref(), rows.task.as_ref()); + let first = |values: [Option; 2]| values.into_iter().flatten().next(); + let time = |values: [Option>; 2]| values.into_iter().flatten().next().map(micros); + Ok(json!({ + "can_continue": can_continue, + "cancel_requested": cancel_requested, + "reason": reason, + "heartbeat_recorded": false, + "workflow_instance_id": text(run, "workflow_instance_id")?, + "workflow_run_id": first([text(run, "id")?, first([text(activity, "workflow_run_id")?, text(attempt, "workflow_run_id")?])]), + "workflow_task_id": first([text(task, "id")?, text(attempt, "workflow_task_id")?]), + "activity_execution_id": first([text(activity, "id")?, text(attempt, "activity_execution_id")?]), + "activity_attempt_id": attempt_id, + "attempt_number": DB::number(&rows.attempt, "attempt_number").ok(), + "run_status": text(run, "status")?, + "run_closed_reason": text(run, "closed_reason")?, + "run_closed_at": time([instant(run, "closed_at")?, None]), + "activity_status": text(activity, "status")?, + "attempt_status": text(attempt, "status")?, + "task_status": text(task, "status")?, + "lease_owner": first([text(attempt, "lease_owner")?, text(task, "lease_owner")?]), + "lease_expires_at": time([instant(attempt, "lease_expires_at")?, instant(task, "lease_expires_at")?]), + "last_heartbeat_at": time([instant(attempt, "last_heartbeat_at")?, instant(activity, "last_heartbeat_at")?]), + "heartbeat_deadline_at": time([instant(activity, "heartbeat_deadline_at")?, None]), + })) + } + + /// Workflow's `attemptContinuationState`: whether the attempt may keep + /// running, whether it was asked to stop, and why not. + fn continuation(rows: &Rows, at: DateTime) -> Result<(bool, bool, Option<&'static str>)> { + let attempt_status = DB::string(&rows.attempt, "status")?; + let run_status = match &rows.run { + Some(run) => Some(DB::string(run, "status")?), + None => None, + }; + if attempt_status != "running" { + return Ok(match run_status.as_deref() { + Some("cancelled") => (false, true, Some("run_cancelled")), + Some("terminated") => (false, true, Some("run_terminated")), + _ => (false, attempt_status == "cancelled", Some("attempt_closed")), + }); + } + let Some(activity) = &rows.activity else { + return Ok((false, false, Some("activity_execution_missing"))); + }; + let (Some(run), Some(run_status)) = (&rows.run, run_status) else { + return Ok((false, false, Some("workflow_run_missing"))); + }; + let activity_status = DB::string(activity, "status")?; + let task_status = match &rows.task { + Some(task) => Some(DB::string(task, "status")?), + None => None, + }; + match run_status.as_str() { + "cancelled" => return Ok((false, true, Some("run_cancelled"))), + "terminated" => return Ok((false, true, Some("run_terminated"))), + _ => {} + } + if activity_status == "cancelled" { + return Ok((false, true, Some("activity_cancelled"))); + } + if task_status.as_deref() == Some("cancelled") { + return Ok((false, true, Some("task_cancelled"))); + } + if !matches!(run_status.as_str(), "pending" | "running" | "waiting") { + return Ok((false, false, Some("run_closed"))); + } + if activity_status != "running" { + return Ok((false, false, Some("activity_not_running"))); + } + if DB::optional_string(activity, "current_attempt_id")?.as_deref() != Some(DB::string(&rows.attempt, "id")?.as_str()) + || DB::number(activity, "attempt_count")? != DB::number(&rows.attempt, "attempt_number")? + { + return Ok((false, false, Some("stale_attempt"))); + } + let Some(task) = &rows.task else { + return Ok((false, false, Some("task_not_found"))); + }; + let payload = DB::optional_document_row(task, "payload")?.unwrap_or(Value::Null); + if DB::string(task, "workflow_run_id")? != DB::string(activity, "workflow_run_id")? + || payload["activity_execution_id"].as_str() != Some(DB::string(activity, "id")?.as_str()) + || DB::number(task, "attempt_count")? != DB::number(&rows.attempt, "attempt_number")? + { + return Ok((false, false, Some("stale_task"))); + } + if task_status.as_deref() != Some("leased") { + return Ok((false, false, Some("task_not_leased"))); + } + // `ActivityAttemptAuthority::expiredReason`: the task's lease, then + // the activity's business deadlines, then the run's. + if DB::optional_instant(task, "lease_expires_at")?.is_none_or(|expires| at >= expires) { + return Ok((false, false, Some("lease_expired"))); + } + if let Some(kind) = super::activity_deadlines::expired_kind::(activity, at)? { + return Ok((false, false, Some(super::activity_deadlines::timeout_reason(kind)))); + } + for (reason, column) in [("run_timeout", "run_deadline_at"), ("execution_timeout", "execution_deadline_at")] { + if DB::optional_instant(run, column)?.is_some_and(|deadline| at >= deadline) { + return Ok((false, false, Some(reason))); + } + } + Ok((true, false, None)) + } +} diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index e0d2a37ad..5f3465bba 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -1130,6 +1130,7 @@ impl Runtime { } delegate!(heartbeat_activity(task_id: &str, body: Value, protocol: &str) -> Result); delegate!(activity_status(task_id: &str, body: Value, protocol: &str) -> Result); + delegate!(renew_activity_lease(task_id: &str, body: Value) -> Result); pub(crate) async fn fail_activity(&self, task_id: &str, body: Value) -> Result { let (failure, blob) = super::activity_failures::prepare_failure(&body, self.signal_codec.clone()).await?; diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 40999546b..612d0afb4 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -260,6 +260,10 @@ pub fn router(runtime: Runtime) -> Router { "/api/worker/activity-tasks/{task_id}/heartbeat", post(heartbeat_activity), ) + .route( + "/api/worker/activity-tasks/{task_id}/renew-lease", + post(renew_activity_lease), + ) .route( "/api/worker/activity-tasks/{task_id}/fail", post(fail_activity), @@ -698,7 +702,7 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { | [ "activity-tasks", _, - "complete" | "fail" | "heartbeat" | "status" + "complete" | "fail" | "heartbeat" | "status" | "renew-lease" ] ); } @@ -1750,6 +1754,14 @@ async fn heartbeat_activity( .map(Json) } +async fn renew_activity_lease( + Extension(runtime): Extension, + Path(task_id): Path, + Json(body): Json, +) -> Result> { + runtime.renew_activity_lease(&task_id, body).await.map(Json) +} + async fn activity_status( Extension(runtime): Extension, Path(task_id): Path, diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index a4dc48bf7..bb84f11bc 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -2,11 +2,33 @@ //! arguments use the official codec on a bounded blocking worker. //! Typed database adapters share transitions; existing PHP data requires qualification. +/// Store methods for every backend: the bounds the shared SQL helpers need, +/// written once rather than in each module. +macro_rules! store_methods { + ($($body:tt)*) => { + impl $crate::runtime::store::Store + where + for<'c> &'c mut DB::Connection: ::sqlx::Executor<'c, Database = DB>, + DB::Arguments: ::sqlx::IntoArguments, + for<'q> &'q str: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'q> Option<&'q str>: ::sqlx::Encode<'q, DB>, + for<'q> Option: ::sqlx::Encode<'q, DB>, + for<'q> String: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'q> i64: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'r> i64: ::sqlx::Decode<'r, DB>, + usize: ::sqlx::ColumnIndex, + { + $($body)* + } + }; +} + mod active_summary; mod activity_completion_uploads; mod activity_deadlines; mod activity_failures; mod activity_heartbeats; +mod activity_lease_renewal; mod activity_metadata; mod activity_metadata_read; mod activity_results; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 9d7cc8166..b798a252a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1589,7 +1589,7 @@ where super::activity_deadlines::validate_policy(&policy)?; let attempt_id = id(); let started_at = now(); - Self::query("UPDATE activity_attempts SET status='expired',closed_at=$1 WHERE activity_execution_id=$2 AND status='running'") + Self::query("UPDATE activity_attempts SET status='expired',lease_expires_at=NULL,closed_at=$1 WHERE activity_execution_id=$2 AND status='running'") .bind(DB::bind_time(started_at)).bind(&activity_id).execute(&mut *tx).await?; Self::query("INSERT INTO activity_attempts(id,activity_execution_id,workflow_run_id,workflow_task_id,attempt_number,status,lease_owner,lease_expires_at,started_at) VALUES ($1,$2,$3,$4,$5,'running',$6,$7,$8)") .bind(&attempt_id).bind(&activity_id).bind(&run_id).bind(&task_id).bind(attempt) @@ -2808,12 +2808,12 @@ where "ActivityCompleted", ) .await?; - Self::query("UPDATE activity_attempts SET status='completed',closed_at=$1 WHERE id=$2") + Self::query("UPDATE activity_attempts SET status='completed',lease_expires_at=NULL,closed_at=$1 WHERE id=$2") .bind(DB::bind_time(now())) .bind(attempt_id) .execute(&mut *tx) .await?; - Self::query("UPDATE workflow_tasks SET status='completed' WHERE id=$1") + Self::query("UPDATE workflow_tasks SET status='completed',lease_expires_at=NULL WHERE id=$1") .bind(task_id) .execute(&mut *tx) .await?; @@ -3816,7 +3816,8 @@ pub(crate) fn capabilities() -> Value { "worker_deregistration_fencing":{"schema":"durable-workflow.v2.worker-deregistration.v1","supported":true,"receipt_retention_seconds":600,"endpoint":"/worker/registrations/{workerId}/deregister"}, "workflow_memo_updates": false, "cooperative_cancellation": true, "prepared_local_activities": false, "worker_sessions": false, "sticky_execution": false, "local_activities": false, "message_streams": false, - "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true}); + "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true, + "activity_lease_renewal": true}); value["supported_workflow_task_commands"] .as_array_mut() .expect("commands") diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index a1bc26b10..2f60a1c43 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -130,6 +130,9 @@ mod worker_pressure; #[path = "execution/version_markers.rs"] mod version_markers; +#[path = "execution/activity_lease_renewal.rs"] +mod activity_lease_renewal; + #[path = "execution/activity_deadlines.rs"] mod activity_deadlines; diff --git a/rust/tests/execution/activity_lease_renewal.rs b/rust/tests/execution/activity_lease_renewal.rs new file mode 100644 index 000000000..85849e2b0 --- /dev/null +++ b/rust/tests/execution/activity_lease_renewal.rs @@ -0,0 +1,173 @@ +use super::*; + +// Expectations measured against published Server 2.11.1 (Workflow 2.10.2): +// POST /api/worker/activity-tasks/{task}/renew-lease. + +fn lease_activity() -> Value { + json!({"type":"schedule_activity","activity_type":"lease_activity", + "arguments":envelope(Payload::Array(vec![])), + "retry_policy":{"max_attempts":1,"backoff_seconds":[]}, + "heartbeat_timeout":60,"start_to_close_timeout":300}) +} + +/// Starts a run and claims its single activity. +async fn claim(app: &Router, workflow_id: &str) -> (Value, Value, Value) { + let started = start(app, workflow_id).await; + let workflow = poll(app, "lease-worker", "workflow").await; + assert_eq!(workflow["run_id"], started["run_id"], "{workflow}"); + assert_eq!(finish_task(app, &workflow, json!([lease_activity()])).await.0, StatusCode::OK); + let activity = poll(app, "lease-worker", "activity").await; + assert_eq!(activity["run_id"], started["run_id"], "{activity}"); + (started, workflow, activity) +} + +fn own(activity: &Value) -> Value { + json!({"activity_attempt_id":activity["activity_attempt_id"],"lease_owner":activity["lease_owner"]}) +} + +fn path(activity: &Value, verb: &str) -> String { + format!("/api/worker/activity-tasks/{}/{verb}", activity["task_id"].as_str().unwrap()) +} + +async fn renew(app: &Router, activity: &Value, body: Value) -> (StatusCode, Value) { + request_protocol(app, "POST", &path(activity, "renew-lease"), body, "1.20").await +} + +fn without_protocol(mut body: Value) -> Value { + let map = body.as_object_mut().unwrap(); + assert_eq!(map.remove("protocol_version"), Some(json!("1.20"))); + assert_eq!(map.remove("server_capabilities").unwrap()["activity_lease_renewal"], true); + body +} + +#[tokio::test] +async fn renewal_extends_only_the_live_lease() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "lease-worker", json!(["echo"]), json!(["lease_activity"])).await; + let (started, workflow, activity) = claim(&app, "lease-live").await; + let status_before = request_protocol(&app, "POST", &path(&activity, "status"), own(&activity), "1.20").await; + assert_eq!(status_before.0, StatusCode::OK, "{}", status_before.1); + let history_before = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + let (code, renewed) = renew(&app, &activity, own(&activity)).await; + assert_eq!(code, StatusCode::OK, "{renewed}"); + let mut renewed = without_protocol(renewed); + let lease = renewed.as_object_mut().unwrap().remove("lease_expires_at").unwrap(); + let server_time = renewed.as_object_mut().unwrap().remove("server_time").unwrap(); + let heartbeat_deadline = renewed.as_object_mut().unwrap().remove("heartbeat_deadline_at").unwrap(); + assert_eq!(renewed, json!({ + "can_continue":true,"cancel_requested":false,"reason":null,"heartbeat_recorded":false, + "workflow_instance_id":"lease-live","workflow_run_id":started["run_id"],"workflow_task_id":activity["task_id"], + "activity_execution_id":activity["activity_execution_id"],"activity_attempt_id":activity["activity_attempt_id"], + "attempt_number":1,"run_status":"waiting","run_closed_reason":null,"run_closed_at":null, + "activity_status":"running","attempt_status":"running","task_status":"leased", + "lease_owner":activity["lease_owner"],"last_heartbeat_at":null, + "lease_renewed":true,"task_id":activity["task_id"],"worker_session":null})); + let micros = |value: &Value| { + let text = value.as_str().unwrap(); + assert!(text.ends_with('Z') && text.len() == 27, "{text}"); + chrono::DateTime::parse_from_rfc3339(text).unwrap() + }; + let time = micros(&server_time); + assert_eq!(micros(&lease), time + chrono::Duration::seconds(300), "the lease runs the configured 300 s from now"); + assert!(lease.as_str().unwrap() > status_before.1["lease_expires_at"].as_str().unwrap()); + assert_eq!(heartbeat_deadline, status_before.1["deadlines"]["heartbeat"], "progress deadlines never move"); + + let status_after = request_protocol(&app, "POST", &path(&activity, "status"), own(&activity), "1.20").await; + let instant = |value: &Value| chrono::DateTime::parse_from_rfc3339(value.as_str().unwrap()).unwrap(); + assert_eq!(instant(&status_after.1["lease_expires_at"]), instant(&lease), "status reads the renewed lease"); + assert_eq!(status_after.1["last_heartbeat_at"], Value::Null); + assert_eq!(status_after.1["deadlines"], status_before.1["deadlines"]); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history_before, + "renewal adds no history"); + + // Like PHP, a 1.19 request and an unruled field are accepted. + assert_eq!(request(&app, "POST", &path(&activity, "renew-lease"), own(&activity)).await.0, StatusCode::OK); + let mut extra = own(&activity); + extra["message"] = json!("progress"); + assert_eq!(renew(&app, &activity, extra).await.0, StatusCode::OK); + + // Ownership refusals, in PHP's order and shape. + let refusal = |task: &Value, attempt: &Value, error: &str, reason: &str| { + json!({"task_id":task,"activity_attempt_id":attempt,"error":error,"reason":reason}) + }; + let (code, body) = renew(&app, &activity, json!({"activity_attempt_id":activity["activity_attempt_id"],"lease_owner":"someone-else"})).await; + let mut expected = refusal(&activity["task_id"], &activity["activity_attempt_id"], + "Activity attempt lease is owned by another worker.", "lease_owner_mismatch"); + expected["lease_owner"] = activity["lease_owner"].clone(); + assert_eq!((code, without_protocol(body)), (StatusCode::CONFLICT, expected)); + let (code, body) = renew(&app, &activity, json!({"activity_attempt_id":"no-such-attempt","lease_owner":activity["lease_owner"]})).await; + assert_eq!((code, without_protocol(body)), (StatusCode::NOT_FOUND, + refusal(&activity["task_id"], &json!("no-such-attempt"), "Activity attempt not found.", "attempt_not_found"))); + let (code, body) = request_protocol(&app, "POST", "/api/worker/activity-tasks/no-such-task/renew-lease", own(&activity), "1.20").await; + assert_eq!((code, without_protocol(body)), (StatusCode::NOT_FOUND, + refusal(&json!("no-such-task"), &activity["activity_attempt_id"], "Activity task not found.", "task_not_found"))); + // Another task of this namespace, here the run's workflow task. + let (code, body) = request_protocol(&app, "POST", &path(&workflow, "renew-lease"), own(&activity), "1.20").await; + assert_eq!((code, without_protocol(body)), (StatusCode::CONFLICT, + refusal(&workflow["task_id"], &activity["activity_attempt_id"], "Activity attempt is not leased for this task.", "task_mismatch"))); + + // Laravel's validation. + let (code, body) = renew(&app, &activity, json!({})).await; + assert_eq!((code, without_protocol(body)), (StatusCode::UNPROCESSABLE_ENTITY, json!({ + "message":"The activity attempt id field is required. (and 1 more error)","reason":"validation_failed", + "errors":{"activity_attempt_id":["The activity attempt id field is required."],"lease_owner":["The lease owner field is required."]}, + "validation_errors":{"activity_attempt_id":["The activity attempt id field is required."],"lease_owner":["The lease owner field is required."]}}))); + let (code, body) = renew(&app, &activity, json!({"activity_attempt_id":5,"lease_owner":["x"]})).await; + assert_eq!(code, StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!(body["errors"], json!({"activity_attempt_id":["The activity attempt id field must be a string."], + "lease_owner":["The lease owner field must be a string."]})); + + // A recorded heartbeat is reported; renewal still records none. + let beat = request_protocol(&app, "POST", &path(&activity, "heartbeat"), own(&activity), "1.20").await; + assert_eq!(beat.1["heartbeat_recorded"], true, "{}", beat.1); + let (_, after_beat) = renew(&app, &activity, own(&activity)).await; + assert_eq!(instant(&after_beat["last_heartbeat_at"]), instant(&beat.1["last_heartbeat_at"])); + assert_ne!(after_beat["heartbeat_deadline_at"], heartbeat_deadline, "the heartbeat moved its own deadline"); + + // A closed attempt reports its status and cannot be renewed. + let completed = request_protocol(&app, "POST", &path(&activity, "complete"), + json!({"activity_attempt_id":activity["activity_attempt_id"],"lease_owner":activity["lease_owner"], + "result":envelope(Payload::String("done".into()))}), "1.20").await; + assert_eq!(completed.0, StatusCode::OK, "{}", completed.1); + let (code, closed) = renew(&app, &activity, own(&activity)).await; + assert_eq!(code, StatusCode::CONFLICT, "{closed}"); + let closed = without_protocol(closed); + assert_eq!([&closed["can_continue"], &closed["cancel_requested"], &closed["reason"], &closed["lease_renewed"], + &closed["activity_status"], &closed["attempt_status"], &closed["task_status"], &closed["lease_expires_at"]], + [&json!(false), &json!(false), &json!("attempt_closed"), &json!(false), &json!("completed"), + &json!("completed"), &json!("completed"), &Value::Null]); + for absent in ["task_id", "server_time", "worker_session"] { + assert!(closed.get(absent).is_none(), "a refused status carries no {absent}: {closed}"); + } +} + +#[tokio::test] +async fn renewal_cannot_revive_an_expired_or_closed_run() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "lease-worker", json!(["echo"]), json!(["lease_activity"])).await; + let (started, _, live) = claim(&app, "lease-terminated").await; + let terminated = request(&app, "POST", &format!("/api/workflows/{}/terminate", started["workflow_id"].as_str().unwrap()), + json!({"reason":"measure"})).await; + assert_eq!(terminated.0, StatusCode::OK, "{}", terminated.1); + let (code, body) = renew(&app, &live, own(&live)).await; + assert_eq!(code, StatusCode::CONFLICT, "{body}"); + assert_eq!([&body["can_continue"], &body["cancel_requested"], &body["reason"], &body["run_status"], + &body["run_closed_reason"], &body["attempt_status"], &body["lease_renewed"]], + [&json!(false), &json!(true), &json!("run_terminated"), &json!("terminated"), &json!("terminated"), + &json!("cancelled"), &json!(false)]); + // Last: an expired lease is redelivered as a new attempt. + let (_, _, activity) = claim(&app, "lease-expired").await; + database.execute("UPDATE workflow_tasks SET lease_expires_at='2020-01-02 03:04:05.123456' WHERE task_type='activity'").await; + let (code, expired) = renew(&app, &activity, own(&activity)).await; + assert_eq!(code, StatusCode::CONFLICT, "{expired}"); + assert_eq!([&expired["reason"], &expired["lease_renewed"], &expired["can_continue"]], + [&json!("lease_expired"), &json!(false), &json!(false)]); + let status = request_protocol(&app, "POST", &path(&activity, "status"), own(&activity), "1.20").await; + assert_eq!(status.1["can_continue"], false, "an expired lease stays expired: {}", status.1); +} diff --git a/scripts/conformance/server-parity/activity-lease-renewal-contract.mjs b/scripts/conformance/server-parity/activity-lease-renewal-contract.mjs new file mode 100644 index 000000000..5c7754b54 --- /dev/null +++ b/scripts/conformance/server-parity/activity-lease-renewal-contract.mjs @@ -0,0 +1,113 @@ +import assert from 'node:assert/strict'; + +const MICROS = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$/; +const RENEWALS = ['renewed', 'renewed_on_protocol_1_19', 'extra_field', 'foreign_owner', 'unknown_attempt', + 'unknown_task', 'task_mismatch', 'missing_fields', 'non_string_fields', 'renewed_after_heartbeat', + 'renewed_after_completion', 'renewed_after_cancel', 'renewed_after_terminate']; + +// Milliseconds since the epoch of a UTC timestamp with any fraction length. +function at(value, label) { + assert.ok(typeof value === 'string', `${label}: a timestamp, got ${JSON.stringify(value)}`); + const millis = Date.parse(value.replace(/\.(\d{3})\d*Z$/, '.$1Z')); + assert.ok(!Number.isNaN(millis), `${label}: ${value}`); + return millis; +} + +// Reviewed runtime activity lease renewal (ActivityTaskController::renewLease +// over Workflow's ActivityLeaseRenewal). Identities, times and leases come +// from the observation; their relations and every other field are exact. +export function checkActivityLeaseRenewal(fixture, observation, workflowId) { + const definition = fixture.activity_lease_renewal; + const state = observation.activity_lease_renewal; + const projection = {scope: 'activity_lease_renewal', lease_seconds: definition.lease_seconds}; + assert.ok(state && typeof state === 'object', 'actual lease renewal observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded lease renewal explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const live = state.live; + equal([state.applicable, state.namespace, live.workflow_id], [true, `${workflowId}-lease`.toLowerCase(), `${workflowId}-live`], + 'renewal namespace and run'); + for (const key of RENEWALS) { + equal([state[key]?.response?.protocol_version, state[key]?.lease_renewal_capability], ['1.20', true], + `${key} answers in worker protocol 1.20 and advertises renewal`); + } + // Workflow's attemptStatus for a claim, before any field is overridden. + const attempt = (claim, fields = {}) => ({can_continue: true, cancel_requested: false, reason: null, heartbeat_recorded: false, + workflow_instance_id: claim.workflow_id, workflow_run_id: claim.run_id, workflow_task_id: claim.task_id, + activity_execution_id: claim.activity_execution_id, activity_attempt_id: claim.activity_attempt_id, attempt_number: 1, + run_status: 'waiting', run_closed_reason: null, run_closed_at: null, activity_status: 'running', attempt_status: 'running', + task_status: 'leased', lease_owner: claim.lease_owner, ...fields}); + + // A live renewal: only the lease moves. + const before = state.status_before.response; + equal([state.status_before.status, before.can_continue], [200, true], 'the claim is live'); + equal(state.renewed.status, 200, 'renewal accepted'); + const {lease_expires_at: lease, server_time: serverTime, heartbeat_deadline_at: heartbeatDeadline, protocol_version: _p, + ...renewed} = state.renewed.response; + equal(renewed, {...attempt(live), last_heartbeat_at: null, lease_renewed: true, task_id: live.task_id, worker_session: null}, + 'renewal status'); + assert.match(lease, MICROS, 'lease in UTC microseconds'); + assert.match(serverTime, MICROS, 'server time in UTC microseconds'); + const window = (at(lease, 'lease') - at(serverTime, 'server time')) / 1000; + assert.ok(window > definition.lease_seconds - 1 && window <= definition.lease_seconds, + `the lease runs ${definition.lease_seconds} s from the renewal, not ${window}`); + assert.ok(at(lease, 'lease') > at(before.lease_expires_at, 'claim lease'), 'renewal extends the claim lease'); + equal(heartbeatDeadline, before.deadlines.heartbeat, 'renewal leaves the progress deadline'); + equal(at(state.status_after.response.lease_expires_at, 'status lease'), at(lease, 'lease'), 'status reads the renewed lease'); + equal(state.status_after.response.last_heartbeat_at, null, 'renewal records no heartbeat'); + equal(state.history_after, state.history_before, 'renewal adds no history'); + for (const key of ['renewed_on_protocol_1_19', 'extra_field']) { + equal([state[key].status, state[key].response.lease_renewed], [200, true], `${key} renews`); + } + + // Ownership, in PHP's order and shape, and Laravel's validation. + const refusal = (status, task, attempt, error, reason, extra = {}) => ({status, lease_renewal_capability: true, + response: {task_id: task, activity_attempt_id: attempt, error, reason, ...extra, protocol_version: '1.20'}}); + equal(state.foreign_owner, refusal(409, live.task_id, live.activity_attempt_id, 'Activity attempt lease is owned by another worker.', + 'lease_owner_mismatch', {lease_owner: live.lease_owner}), 'another owner'); + equal(state.unknown_attempt, refusal(404, live.task_id, 'no-such-attempt', 'Activity attempt not found.', 'attempt_not_found'), + 'unknown attempt'); + equal(state.unknown_task, refusal(404, 'no-such-task', live.activity_attempt_id, 'Activity task not found.', 'task_not_found'), + 'unknown task'); + equal(state.task_mismatch, refusal(409, live.workflow_task_id, live.activity_attempt_id, + 'Activity attempt is not leased for this task.', 'task_mismatch'), 'another task of the namespace'); + for (const key of ['missing_fields', 'non_string_fields']) { + equal(state[key], {status: 422, lease_renewal_capability: true, + response: {...definition.validation[key], protocol_version: '1.20'}}, key); + } + + // A heartbeat's progress is reported; renewal keeps its deadline. + equal([state.heartbeat.status, state.heartbeat.response.heartbeat_recorded], [200, true], 'heartbeat recorded'); + const beat = at(state.heartbeat.response.last_heartbeat_at, 'heartbeat'); + const afterBeat = state.renewed_after_heartbeat.response; + equal([state.renewed_after_heartbeat.status, afterBeat.lease_renewed, afterBeat.heartbeat_recorded], [200, true, false], + 'renewal after a heartbeat'); + equal(at(afterBeat.last_heartbeat_at, 'reported heartbeat'), beat, 'the recorded heartbeat is reported'); + equal(at(afterBeat.heartbeat_deadline_at, 'heartbeat deadline') - beat, definition.heartbeat_timeout * 1000, + 'the heartbeat keeps its own deadline'); + + // Closed attempts are refused with their status and no lease. + equal(state.completed.status, 200, 'activity completed'); + const {last_heartbeat_at: closedBeat, heartbeat_deadline_at: closedDeadline, protocol_version: _c, ...closed} = + state.renewed_after_completion.response; + equal(state.renewed_after_completion.status, 409, 'a completed attempt is refused'); + equal(closed, {...attempt(live, {can_continue: false, reason: 'attempt_closed', activity_status: 'completed', + attempt_status: 'completed', task_status: 'completed'}), lease_expires_at: null, lease_renewed: false}, 'completed attempt status'); + equal([at(closedBeat, 'closed heartbeat'), closedDeadline], [beat, afterBeat.heartbeat_deadline_at], 'closed attempt keeps its heartbeat'); + const stopped = (key, claim, command, runStatus, reason) => { + equal(state[command], 200, `${command} accepted`); + const {run_closed_at: closedAt, heartbeat_deadline_at: deadline, protocol_version: _s, ...body} = state[key].response; + equal(state[key].status, 409, `${key} refused`); + assert.match(closedAt, MICROS, `${key} close time in UTC microseconds`); + at(deadline, `${key} heartbeat deadline`); + const {run_closed_at: _r, ...expected} = attempt(claim, {can_continue: false, cancel_requested: true, reason, run_status: runStatus, + run_closed_reason: runStatus, activity_status: 'cancelled', attempt_status: 'cancelled', task_status: 'cancelled'}); + equal(body, {...expected, lease_expires_at: null, last_heartbeat_at: null, lease_renewed: false}, `${key} status`); + }; + stopped('renewed_after_cancel', state.cancelled_claim, 'cancel', 'cancelled', 'run_cancelled'); + stopped('renewed_after_terminate', state.terminated_claim, 'terminate', 'terminated', 'run_terminated'); + return projection; +} diff --git a/scripts/conformance/server-parity/activity-lease-renewal-probe.php b/scripts/conformance/server-parity/activity-lease-renewal-probe.php new file mode 100644 index 000000000..08a9bf8b0 --- /dev/null +++ b/scripts/conformance/server-parity/activity-lease-renewal-probe.php @@ -0,0 +1,111 @@ +createNamespace($namespace, 'activity lease renewal λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $http = new LeaseRenewalHttp(['base_uri' => rtrim($url, '/'), 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30]); + $raw = static function (string $method, string $path, ?array $body = [], string $protocol = '1.20') use ($http, $token, $namespace): array { + $options = ['headers' => array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => $protocol, + ])]; + if ($body !== null) { + $options['json'] = $body === [] ? (object) [] : $body; + } + $response = $http->request($method, $path, $options); + $decoded = json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR); + $capability = is_array($decoded) && array_key_exists('server_capabilities', $decoded) + ? ($decoded['server_capabilities']['activity_lease_renewal'] ?? null) : 'absent'; + unset($decoded['server_capabilities'], $decoded['control_plane']); + + return ['status' => $response->getStatusCode(), 'response' => $decoded, 'lease_renewal_capability' => $capability]; + }; + $ok = static function (array $recorded, int $status, string $operation): array { + if ($recorded['status'] !== $status) { + throw new RuntimeException('Activity lease renewal '.$operation.': '.json_encode($recorded)); + } + + return $recorded['response']; + }; + // Avro payloads precomputed by the published serializer: [] and 'done'. + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $done = ['codec' => 'avro', 'blob' => $definition['result_blob']]; + // Starts a run on its own queue with its own worker and claims its single + // activity. + $claim = static function (string $suffix) use ($raw, $ok, $empty, $definition, $workflowId): array { + $id = $workflowId.'-'.$suffix; + $queue = $id.'-queue'; + $worker = $id.'-worker'; + $ok($raw('POST', '/api/worker/register', ['worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'php', + 'sdk_version' => 'raw-http', 'supported_workflow_types' => [$definition['workflow_type']], + 'supported_activity_types' => [$definition['activity_type']], 'capabilities' => [], + 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http'])]), 201, 'registration'); + $poll = ['worker_id' => $worker, 'task_queue' => $queue, 'timeout_seconds' => 0]; + $started = $ok($raw('POST', '/api/workflows', ['workflow_id' => $id, 'workflow_type' => $definition['workflow_type'], + 'task_queue' => $queue, 'input' => $empty]), 201, 'start'); + $task = $ok($raw('POST', '/api/worker/workflow-tasks/poll', $poll), 200, 'workflow claim')['task']; + $ok($raw('POST', '/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], + 'commands' => [['type' => 'schedule_activity', 'activity_type' => $definition['activity_type'], + 'arguments' => $empty, 'retry_policy' => ['max_attempts' => 1, 'backoff_seconds' => []], + 'heartbeat_timeout' => $definition['heartbeat_timeout'], 'start_to_close_timeout' => 300]]]), 200, 'schedule activity'); + $activity = $ok($raw('POST', '/api/worker/activity-tasks/poll', $poll), 200, 'activity claim')['task']; + + return ['workflow_id' => $id, 'run_id' => $started['run_id'], 'workflow_task_id' => $task['task_id'], + 'task_id' => $activity['task_id'], 'activity_attempt_id' => $activity['activity_attempt_id'], + 'activity_execution_id' => $activity['activity_execution_id'] ?? null, 'lease_owner' => $activity['lease_owner']]; + }; + $own = static fn (array $claimed): array => ['activity_attempt_id' => $claimed['activity_attempt_id'], 'lease_owner' => $claimed['lease_owner']]; + $renew = static fn (array $claimed, ?array $body = null, string $protocol = '1.20'): array => $raw('POST', + '/api/worker/activity-tasks/'.$claimed['task_id'].'/renew-lease', $body ?? $own($claimed), $protocol); + $status = static fn (array $claimed): array => $raw('POST', '/api/worker/activity-tasks/'.$claimed['task_id'].'/status', $own($claimed)); + $history = static fn (array $claimed): array => array_column($ok($raw('GET', + '/api/workflows/'.$claimed['workflow_id'].'/runs/'.$claimed['run_id'].'/history', null), 200, 'history')['events'] ?? [], 'event_type'); + + $state = ['applicable' => true, 'namespace' => $namespace]; + $live = $state['live'] = $claim('live'); + $state['status_before'] = $status($live); + $state['history_before'] = $history($live); + usleep(20000); + $state['renewed'] = $renew($live); + $state['status_after'] = $status($live); + $state['history_after'] = $history($live); + $state['renewed_on_protocol_1_19'] = $renew($live, null, '1.19'); + $state['extra_field'] = $renew($live, $own($live) + ['message' => 'progress']); + $state['foreign_owner'] = $renew($live, ['activity_attempt_id' => $live['activity_attempt_id'], 'lease_owner' => 'someone-else']); + $state['unknown_attempt'] = $renew($live, ['activity_attempt_id' => 'no-such-attempt', 'lease_owner' => $live['lease_owner']]); + $state['unknown_task'] = $raw('POST', '/api/worker/activity-tasks/no-such-task/renew-lease', $own($live)); + $state['task_mismatch'] = $raw('POST', '/api/worker/activity-tasks/'.$live['workflow_task_id'].'/renew-lease', $own($live)); + $state['missing_fields'] = $renew($live, []); + $state['non_string_fields'] = $renew($live, ['activity_attempt_id' => 5, 'lease_owner' => ['x']]); + $state['heartbeat'] = $raw('POST', '/api/worker/activity-tasks/'.$live['task_id'].'/heartbeat', $own($live) + ['message' => 'half']); + $state['renewed_after_heartbeat'] = $renew($live); + $state['completed'] = $raw('POST', '/api/worker/activity-tasks/'.$live['task_id'].'/complete', $own($live) + ['result' => $done]); + $state['renewed_after_completion'] = $renew($live); + + $cancelled = $state['cancelled_claim'] = $claim('cancelled'); + $state['cancel'] = $raw('POST', '/api/workflows/'.$cancelled['workflow_id'].'/cancel', ['reason' => 'lease renewal fixture'])['status']; + $state['renewed_after_cancel'] = $renew($cancelled); + $terminated = $state['terminated_claim'] = $claim('terminated'); + $state['terminate'] = $raw('POST', '/api/workflows/'.$terminated['workflow_id'].'/terminate', ['reason' => 'lease renewal fixture'])['status']; + $state['renewed_after_terminate'] = $renew($terminated); + + return $state; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index a181ab10d..afa2a2ac7 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -12,6 +12,7 @@ import {checkWorkerDeregistration} from './worker-deregistration-contract.mjs'; import {checkLateWorkflowCompletion} from './late-workflow-completion-contract.mjs'; import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkStorageTest} from './storage-test-contract.mjs'; +import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; import {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; import {checkSignalsBeforeWorker} from './signals-before-worker-contract.mjs'; @@ -141,6 +142,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const lateWorkflowCompletion = fixture.late_workflow_completion ? checkLateWorkflowCompletion(fixture, observation, workflowId) : null; const searchAttributes = fixture.search_attributes ? checkSearchAttributes(fixture, observation, workflowId) : null; const storageTest = fixture.storage_test ? checkStorageTest(fixture, observation, workflowId) : null; + const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null; const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null; const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId) : null; @@ -687,6 +689,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(lateWorkflowCompletion ? {late_workflow_completion: lateWorkflowCompletion} : {}), ...(searchAttributes ? {search_attributes: searchAttributes} : {}), ...(storageTest ? {storage_test: storageTest} : {}), + ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), ...(signalsBeforeWorker ? {signals_before_worker: signalsBeforeWorker} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 0b40d67a2..0f6c17a7d 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -430,6 +430,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $lateWorkflowCompletion = isset($fixture['late_workflow_completion']) ? finishHttpLateWorkflowCompletion($client, $fixture, $workflowId) : null; $searchAttributes = isset($fixture['search_attributes']) ? finishHttpSearchAttributes($client, $fixture, $workflowId, $url) : null; $storageTest = isset($fixture['storage_test']) ? finishHttpStorageTest($client, $fixture, $workflowId, $url) : null; + $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null; $schedulePolicies = isset($fixture['schedule_policies']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url) : null; $scheduleCatalog = isset($fixture['schedule_catalog']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url, 'schedule_catalog') @@ -529,6 +530,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'late_workflow_completion' => $lateWorkflowCompletion, 'search_attributes' => $searchAttributes, 'storage_test' => $storageTest, + 'activity_lease_renewal' => $activityLeaseRenewal, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, 'signals_before_worker' => $signalsBeforeWorker, @@ -832,6 +834,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/namespace-probe.php'; require __DIR__.'/search-attribute-probe.php'; require __DIR__.'/storage-test-probe.php'; + require __DIR__.'/activity-lease-renewal-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; @@ -990,6 +993,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['storage_test'])) { $observation['storage_test'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_storage_diagnostic']; } + if ($mode === 'embedded' && isset($fixture['activity_lease_renewal'])) { + $observation['activity_lease_renewal'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; + } if ($mode === 'embedded' && isset($fixture['schedule_policies'])) { $observation['schedule_policies'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_schedules']; } diff --git a/tests/Fixtures/ServerParity/activity-lease-renewal.json b/tests/Fixtures/ServerParity/activity-lease-renewal.json new file mode 100644 index 000000000..e6e569ce0 --- /dev/null +++ b/tests/Fixtures/ServerParity/activity-lease-renewal.json @@ -0,0 +1,37 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "activity-lease-renewal", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "lease renewal λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "lease renewal λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "activity_lease_renewal": { + "workflow_type": "parity.v1.lease", + "activity_type": "parity.v1.lease_activity", + "heartbeat_timeout": 60, + "lease_seconds": 300, + "empty_blob": "wwHioz3/VYAiNwwA", + "result_blob": "wwHioz3/VYAiNwoIZG9uZQ==", + "validation": { + "missing_fields": { + "message": "The activity attempt id field is required. (and 1 more error)", + "reason": "validation_failed", + "errors": {"activity_attempt_id": ["The activity attempt id field is required."], "lease_owner": ["The lease owner field is required."]}, + "validation_errors": {"activity_attempt_id": ["The activity attempt id field is required."], "lease_owner": ["The lease owner field is required."]} + }, + "non_string_fields": { + "message": "The activity attempt id field must be a string. (and 1 more error)", + "reason": "validation_failed", + "errors": {"activity_attempt_id": ["The activity attempt id field must be a string."], "lease_owner": ["The lease owner field must be a string."]}, + "validation_errors": {"activity_attempt_id": ["The activity attempt id field must be a string."], "lease_owner": ["The lease owner field must be a string."]} + } + } + }, + "contract": "POST /api/worker/activity-tasks/{task}/renew-lease (published Server 2.11, advertised as server_capabilities.activity_lease_renewal) keeps a live attempt's lease for its owner. It answers 200 with the attempt's status: can_continue true, no heartbeat recorded, the run, task, execution and attempt identities, the stored heartbeat and its deadline unchanged, lease_renewed true, server_time, task_id and worker_session null. The lease then runs the configured 300 seconds from the renewal, never less than before, and the run's history is unchanged. A 1.19 request and an unruled body field are accepted. Ownership is checked in PHP's order with its bodies: an unknown task or attempt is 404, another task or another owner 409. Missing or non-string fields fail Laravel's validation. A heartbeat's progress is reported and its own deadline kept. A completed, cancelled or terminated attempt is refused with its status and lease_renewed false, with no lease. Embedded applications have no worker protocol." +} diff --git a/tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs b/tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs new file mode 100644 index 000000000..eb454b44b --- /dev/null +++ b/tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs @@ -0,0 +1,110 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkActivityLeaseRenewal} from '../../scripts/conformance/server-parity/activity-lease-renewal-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/activity-lease-renewal.json', import.meta.url))); +const definition = fixture.activity_lease_renewal; +const id = 'parity-v1-activity-lease-renewal'; + +// Shapes published Server 2.11.1 returned. +const claim = (suffix, n) => ({workflow_id: `${id}-${suffix}`, run_id: `run-${n}`, workflow_task_id: `wft-${n}`, + task_id: `task-${n}`, activity_attempt_id: `ATTEMPT-${n}`, activity_execution_id: `exec-${n}`, lease_owner: `${id}-${suffix}-worker`}); +const live = claim('live', 1); +const cancelled = claim('cancelled', 2); +const terminated = claim('terminated', 3); +const status = (claimed, fields = {}) => ({can_continue: true, cancel_requested: false, reason: null, heartbeat_recorded: false, + workflow_instance_id: claimed.workflow_id, workflow_run_id: claimed.run_id, workflow_task_id: claimed.task_id, + activity_execution_id: claimed.activity_execution_id, activity_attempt_id: claimed.activity_attempt_id, attempt_number: 1, + run_status: 'waiting', run_closed_reason: null, run_closed_at: null, activity_status: 'running', attempt_status: 'running', + task_status: 'leased', lease_owner: claimed.lease_owner, lease_expires_at: '2026-10-22T20:52:13.554963Z', last_heartbeat_at: null, + heartbeat_deadline_at: '2026-10-22T20:48:11.939526Z', ...fields}); +const reply = (code, response) => ({status: code, response: {...response, protocol_version: '1.20'}, lease_renewal_capability: true}); +const renewed = (fields = {}) => reply(200, {...status(live), lease_renewed: true, server_time: '2026-10-22T20:47:13.592509Z', + task_id: live.task_id, worker_session: null, ...fields}); +const refusal = (code, task, attempt, error, reason, extra = {}) => reply(code, {task_id: task, activity_attempt_id: attempt, error, reason, ...extra}); +const stopped = (claimed, runStatus, reason) => reply(409, {...status(claimed, {can_continue: false, cancel_requested: true, reason, + run_status: runStatus, run_closed_reason: runStatus, run_closed_at: '2026-10-22T20:48:18.603147Z', activity_status: 'cancelled', + attempt_status: 'cancelled', task_status: 'cancelled', lease_expires_at: null}), lease_renewed: false}); +const history = ['StartAccepted', 'WorkflowStarted', 'ActivityScheduled', 'ActivityStarted']; +const observation = () => structuredClone({mode: 'http', activity_lease_renewal: {applicable: true, namespace: `${id}-lease`, live, + status_before: reply(200, {task_id: live.task_id, can_continue: true, lease_expires_at: '2026-10-22T20:52:11.939673Z', + last_heartbeat_at: null, deadlines: {start_to_close: '2026-10-22T20:52:11.939526Z', heartbeat: '2026-10-22T20:48:11.939526Z'}}), + history_before: [...history], renewed: renewed(), + status_after: reply(200, {task_id: live.task_id, can_continue: true, lease_expires_at: '2026-10-22T20:52:13.554963Z', last_heartbeat_at: null}), + history_after: [...history], + renewed_on_protocol_1_19: renewed(), extra_field: renewed(), + foreign_owner: refusal(409, live.task_id, live.activity_attempt_id, 'Activity attempt lease is owned by another worker.', + 'lease_owner_mismatch', {lease_owner: live.lease_owner}), + unknown_attempt: refusal(404, live.task_id, 'no-such-attempt', 'Activity attempt not found.', 'attempt_not_found'), + unknown_task: refusal(404, 'no-such-task', live.activity_attempt_id, 'Activity task not found.', 'task_not_found'), + task_mismatch: refusal(409, live.workflow_task_id, live.activity_attempt_id, 'Activity attempt is not leased for this task.', 'task_mismatch'), + missing_fields: reply(422, definition.validation.missing_fields), + non_string_fields: reply(422, definition.validation.non_string_fields), + heartbeat: reply(200, {task_id: live.task_id, heartbeat_recorded: true, last_heartbeat_at: '2026-10-22T20:47:14.723462Z'}), + renewed_after_heartbeat: renewed({last_heartbeat_at: '2026-10-22T20:47:14.723462Z', heartbeat_deadline_at: '2026-10-22T20:48:14.723462Z'}), + completed: reply(200, {outcome: 'completed', recorded: true}), + renewed_after_completion: reply(409, {...status(live, {can_continue: false, reason: 'attempt_closed', activity_status: 'completed', + attempt_status: 'completed', task_status: 'completed', lease_expires_at: null, last_heartbeat_at: '2026-10-22T20:47:14.723462Z', + heartbeat_deadline_at: '2026-10-22T20:48:14.723462Z'}), lease_renewed: false}), + cancelled_claim: cancelled, cancel: 200, renewed_after_cancel: stopped(cancelled, 'cancelled', 'run_cancelled'), + terminated_claim: terminated, terminate: 200, renewed_after_terminate: stopped(terminated, 'terminated', 'run_terminated')}}); + +test('lease renewal accepts the published shapes', () => { + assert.deepStrictEqual(checkActivityLeaseRenewal(fixture, observation(), id), {scope: 'activity_lease_renewal', lease_seconds: 300}); +}); + +test('lease renewal accepts a native lease exactly 300 s after its millisecond renewal', () => { + const state = observation(); + const renewal = state.activity_lease_renewal.renewed.response; + renewal.server_time = '2026-10-22T20:47:13.592000Z'; + renewal.lease_expires_at = '2026-10-22T20:52:13.592000Z'; + state.activity_lease_renewal.status_after.response.lease_expires_at = '2026-10-22T20:52:13.592Z'; + assert.doesNotThrow(() => checkActivityLeaseRenewal(fixture, state, id)); +}); + +test('embedded lease renewal is explicitly inapplicable', () => { + const projection = checkActivityLeaseRenewal(fixture, + {mode: 'embedded', activity_lease_renewal: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}, id); + assert.deepStrictEqual(projection, {scope: 'activity_lease_renewal', lease_seconds: 300}); + assert.throws(() => checkActivityLeaseRenewal(fixture, {mode: 'embedded', activity_lease_renewal: {applicable: true}}, id)); +}); + +for (const [label, mutate] of [ + ['an unadvertised capability', state => { state.renewed.lease_renewal_capability = null; }], + ['an older protocol version', state => { state.foreign_owner.response.protocol_version = '1.19'; }], + ['a refused live renewal', state => { state.renewed.status = 409; }], + ['a recorded heartbeat', state => { state.renewed.response.heartbeat_recorded = true; }], + ['a missing server time', state => { delete state.renewed.response.server_time; }], + ['a millisecond lease', state => { state.renewed.response.lease_expires_at = '2026-10-22T20:52:13.554Z'; }], + ['a lease shorter than the window', state => { state.renewed.response.lease_expires_at = '2026-10-22T20:51:13.554963Z'; }], + ['a lease longer than the window', state => { state.renewed.response.lease_expires_at = '2026-10-22T20:52:14.554963Z'; }], + ['a moved progress deadline', state => { state.renewed.response.heartbeat_deadline_at = '2026-10-22T20:48:13.592509Z'; }], + ['added history', state => { state.history_after.push('ActivityHeartbeatRecorded'); }], + ['an unrenewed status read', state => { state.status_after.response.lease_expires_at = '2026-10-22T20:52:11.939673Z'; }], + ['a session object', state => { state.renewed.response.worker_session = {}; }], + ['a refused 1.19 request', state => { state.renewed_on_protocol_1_19.status = 422; }], + ['a refused unruled field', state => { state.extra_field.status = 422; }], + ['a foreign owner accepted', state => { state.foreign_owner.status = 200; }], + ['a foreign owner without its owner', state => { delete state.foreign_owner.response.lease_owner; }], + ['an unknown attempt as 409', state => { state.unknown_attempt.status = 409; }], + ['an unknown task message', state => { state.unknown_task.response.error = 'Task not found.'; }], + ['a task mismatch as 404', state => { state.task_mismatch.status = 404; }], + ['a generic validation refusal', state => { state.missing_fields.response = {reason: 'invalid_request', protocol_version: '1.20'}; }], + ['a lost heartbeat', state => { state.renewed_after_heartbeat.response.last_heartbeat_at = null; }], + ['a heartbeat deadline renewed', state => { state.renewed_after_heartbeat.response.heartbeat_deadline_at = '2026-10-22T20:48:15.000000Z'; }], + ['a renewed closed attempt', state => { state.renewed_after_completion.status = 200; }], + ['a closed attempt keeping its lease', state => { state.renewed_after_completion.response.lease_expires_at = '2026-10-22T20:52:14.958720Z'; }], + ['a closed attempt with a server time', state => { state.renewed_after_completion.response.server_time = '2026-10-22T20:47:14.987586Z'; }], + ['a closed attempt with a task id', state => { state.renewed_after_completion.response.task_id = live.task_id; }], + ['a cancellation not requested', state => { state.renewed_after_cancel.response.cancel_requested = false; }], + ['a cancelled attempt keeping its lease', state => { state.renewed_after_cancel.response.lease_expires_at = '2026-10-22T20:52:16.801401Z'; }], + ['a terminated run as cancelled', state => { state.renewed_after_terminate.response.reason = 'run_cancelled'; }], + ['a millisecond close time', state => { state.renewed_after_terminate.response.run_closed_at = '2026-10-22T20:48:18.603Z'; }], +]) { + test(`lease renewal rejects ${label}`, () => { + const value = observation(); + mutate(value.activity_lease_renewal); + assert.throws(() => checkActivityLeaseRenewal(fixture, value, id)); + }); +}