diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index fbf1a5ad9..073d9e4ea 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -1029,6 +1029,21 @@ support is not evidence of multi-node safety. ## Decisions +- **November 2: explicit activity cancellation policies follow Core, except + abandon (#958, #971).** Native 3.0.x refused `cancellation_policy` on + `schedule_activity` and treated every activity as `try_cancel` on whole-run + cooperative delivery. Native now admits `try_cancel` and + `wait_cancellation_completed` under Core 2.15's gate: protocol 1.20 and a + cooperative claim, otherwise 409 `activity_cancellation_policy_not_supported`. + The policy is kept in the activity's options and its `ActivityScheduled` + snapshot. On delivery, `wait_cancellation_completed` holds until the live + attempt's worker acknowledges the stop, and the acknowledgement resumes the + run (`ActivityCancellationWait`). An explicit `abandon` is refused with the + same 409 and `unavailable: ["installed_runtime_activity_policy"]`, because + native has no post-run authority for an abandoned activity yet. It would be + stranded once its run closes. #971 tracks that lifecycle; this is a recorded + difference from published PHP until then. + - **November 2: schedules follow Core 2.15.13's daylight-saving rule, not PHP 2.11.8 (#938).** General fixed daily cron schedules admitting twice when clocks go back in the embedded engine (Workflow #882, Core 2.15.13). diff --git a/rust/src/runtime/activity_cancellation_ack.rs b/rust/src/runtime/activity_cancellation_ack.rs index dfdebeee0..68b3cf10a 100644 --- a/rust/src/runtime/activity_cancellation_ack.rs +++ b/rust/src/runtime/activity_cancellation_ack.rs @@ -155,8 +155,13 @@ store_methods! { let event = Self::query("SELECT h.id FROM workflow_history_events h JOIN workflow_runs r ON r.id=h.workflow_run_id AND r.last_history_sequence=h.sequence WHERE h.workflow_run_id=$1") .bind(&run_id).fetch_one(&mut *tx).await?; let event = DB::string(&event, "id")?; + // A whole-run delivery held for this stop resumes (Server #958). + let resumed = Self::resume_activity_cancellation_wait(&mut tx, &run_id, request_id, &execution_id, &event).await?; tx.summary_changed(&run_id); tx.commit().await?; + if resumed { + self.wake.notify_waiters(); + } Ok(with_protocol(outcome(true, false, None, Some(event)))) } diff --git a/rust/src/runtime/completion_rules.rs b/rust/src/runtime/completion_rules.rs index 567ca3483..8d9be4fc7 100644 --- a/rust/src/runtime/completion_rules.rs +++ b/rust/src/runtime/completion_rules.rs @@ -579,7 +579,7 @@ const FIELD_SCOPES: &[(&str, &[&str], &str)] = &[ ("parent_close_policy", &["start_child_workflow"], "parent_close_policy declares how a child workflow reacts when its parent closes and only applies to a start_child_workflow command."), ("cancellation_policy", &["start_child_workflow", "schedule_activity"], - "cancellation_policy declares how an awaiting workflow handles child or remote activity cancellation. Explicit activity policies require protocol 1.20 and remote abandon requires a finite schedule_to_close_timeout."), + "cancellation_policy declares how an awaiting workflow handles child or remote activity cancellation. Explicit activity policies require protocol 1.20 and a cooperative claim; native admits try_cancel and wait_cancellation_completed, and refuses abandon until abandoned activities can outlive their run."), ("cancellation_scope_id", &["schedule_activity", "start_timer", "start_child_workflow", "open_signal_wait", "open_condition_wait"], "cancellation_scope_id records activity, timer, child or wait membership and requires candidate protocol 1.20."), ("cancellation_cleanup", &["start_timer"], diff --git a/rust/src/runtime/cooperative.rs b/rust/src/runtime/cooperative.rs index 36009dcd2..32f37d166 100644 --- a/rust/src/runtime/cooperative.rs +++ b/rust/src/runtime/cooperative.rs @@ -107,7 +107,8 @@ use super::memo_limits::COMMAND_BATCH_SIZE as PARALLEL_SPAN_LIMIT; /// The reasons `CooperativeCancellationDelivery::validate` returns. /// `DefaultWorkflowTaskBridge::deliverCancellation` answers each with its /// full delivery body under 409: the task and run, no delivery. -const VALIDATION_REASONS: [&str; 8] = [ +const VALIDATION_REASONS: [&str; 9] = [ + "cancellation_activity_abandon_not_supported", "cancellation_request_mismatch", "cancellation_request_history_missing", "cancellation_delivery_mismatch", @@ -221,6 +222,46 @@ where "history_refresh_page_token":"MA=="}))) } + /// The completion guard for an explicit child or activity cancellation + /// policy: protocol 1.20 (strictly `1.N` for an activity) and a + /// cooperative claim, else PHP's 409 naming what is unavailable. + #[allow(clippy::too_many_arguments)] + pub(super) fn cancellation_policy_guard( + task: &DB::Row, + task_id: &str, + body: &Value, + protocol: Option<&str>, + strict_minor: bool, + runtime_gap: Option<&'static str>, + reason: &'static str, + remediation: &'static str, + ) -> Result<()> { + let mut unavailable = Vec::new(); + if !protocol + .and_then(|version| version.strip_prefix("1.")) + .filter(|minor| !strict_minor || !minor.contains('.')) + .and_then(|minor| minor.parse::().ok()) + .is_some_and(|minor| minor >= 20) + { + unavailable.push("request_protocol"); + } + if !Self::claim_supports_cancellation(task)? { + unavailable.push("worker_claim_capability"); + } + unavailable.extend(runtime_gap); + if unavailable.is_empty() { + return Ok(()); + } + Err(RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: json!({"task_id": task_id, "workflow_task_attempt": body["workflow_task_attempt"], + "worker_id": DB::optional_string(task, "lease_owner")?, "outcome": "rejected", "recorded": false, + "reason": reason, "required_capability": "cooperative_cancellation", + "minimum_protocol_version": "1.20", "requested_version": protocol, "unavailable": unavailable, + "remediation": remediation}), + }) + } + pub(super) fn claim_supports_cancellation(task: &DB::Row) -> Result { let payload = DB::document_row(task, "payload")?; let claim = &payload[CLAIM_KEY]; @@ -410,12 +451,15 @@ where "cancellation_delivery_sequence_mismatch", ))); } + // Activities whose `wait_cancellation_completed` policy holds this + // delivery until their workers acknowledge the stop (Server #958). + let mut waits: Vec = Vec::new(); if call_kind == "parallel" { // `parallelEligibility`; a boundary at the next sequence has no // authored member yet. if sequence < next { let members = Self::parallel_members(&mut tx, &run_id, sequence, span, request_sequence).await.map_err(refusal)?; - Self::cancel_members(&mut tx, &run, request_id, task_id, &members).await?; + waits.extend(Self::cancel_members(&mut tx, &run, request_id, task_id, &members).await.map_err(refusal)?); } } else if call_kind == "selection_handle" { // `validateCallBoundary` for a handle: it is awaited at the next @@ -437,7 +481,7 @@ where } vec![(start, kind)] }; - Self::cancel_members(&mut tx, &run, request_id, task_id, &members).await?; + waits.extend(Self::cancel_members(&mut tx, &run, request_id, task_id, &members).await.map_err(refusal)?); } else if call_kind == "child" { if sequence < next { let boundary = Self::validate_child_cancellation_boundary( @@ -467,17 +511,16 @@ where } } } else if call_kind == "activity" { - // `validateCallBoundary`, then `cancelOpenWaitAtSequence` with - // the default `try_cancel` policy: an open activity at the - // sequence is cancelled under the request, and one that resolved - // after the request takes the delivery as it is. Native records - // no explicit activity cancellation policy, so it has no waiting - // or abandoning delivery. + // `validateCallBoundary`, then `cancelOpenWaitAtSequence` under + // the activity's own cancellation policy: an open activity at the + // sequence is cancelled under the request (or left running when + // abandoned), and one that resolved after the request takes the + // delivery as it is. Self::validate_call_boundary(&mut tx, &run_id, sequence, call_kind, sequence < next, request_sequence) .await .map_err(refusal)?; if sequence < next { - Self::deliver_activity_cancellation(&mut tx, &run_id, request_id, sequence, now()).await?; + waits.extend(Self::cancel_activity_member(&mut tx, &run_id, request_id, sequence).await.map_err(refusal)?); } } else if matches!(call_kind, "condition" | "signal") { if sequence < next { @@ -508,6 +551,29 @@ where Self::cancel_root_timer(&mut tx, &run, request_id, &timer, Some(task_id), now()).await?; } } + if !waits.is_empty() { + // `ActivityCancellationWait::prepare`: the delivery waits, the + // claim is released, and the run waits for the stop receipts. The + // task keeps the wait so a receipt can resume it. + let mut held = DB::optional_document_row(&task, "payload")?.unwrap_or_else(|| json!({})); + if !held.is_object() { + held = json!({}); + } + held["cancellation_activity_wait"] = json!({"request_id":request_id, + "root_request_id":requested["cancellation"]["root_request_id"], + "cleanup_deadline_at":requested["cancellation"]["cleanup_deadline_at"],"execution_ids":waits}); + let at = now(); + Self::query("UPDATE workflow_runs SET status='waiting',last_progress_at=$1 WHERE id=$2") + .bind(DB::bind_time(at)).bind(&run_id).execute(&mut *tx).await?; + Self::query("UPDATE workflow_tasks SET status='completed',lease_expires_at=NULL,payload=$1 WHERE id=$2") + .bind(DB::document(&held)).bind(task_id).execute(&mut *tx).await?; + tx.summary_changed(&run_id); + tx.commit().await?; + self.wake.notify_waiters(); + return Ok(json!({"delivered":false,"task_id":task_id,"workflow_run_id":run_id,"request_id":null, + "sequence":null,"call_kind":null,"sequence_span":null,"operation_sequence":null, + "operation_sequence_span":null,"reason":"cancellation_waiting_for_activity","claim_released":true})); + } let at = now(); let expires = self.cancellation_lease_expiry(&run)?; Self::query("UPDATE workflow_tasks SET lease_expires_at=$1 WHERE id=$2") @@ -548,6 +614,144 @@ where Ok(response) } + /// `cancelOpenWaitAtSequence` for an activity under its own + /// `cancellation_policy` (Server #958), `try_cancel` unless its options + /// name another: + /// - `try_cancel` cancels the live attempt and the delivery proceeds; + /// - `wait_cancellation_completed` cancels it too, and returns the + /// execution while its worker has not acknowledged the stop; + /// - `abandon` leaves the activity running, which needs its finite + /// schedule-to-close deadline (`ActivityAbandonment::prepare`). + async fn cancel_activity_member( + tx: &mut Transaction<'_, DB>, + run_id: &str, + request_id: &str, + sequence: i64, + ) -> Result> { + let Some(activity) = Self::query("SELECT id,status,activity_options,schedule_to_close_deadline_at FROM activity_executions WHERE workflow_run_id=$1 AND sequence=$2") + .bind(run_id).bind(sequence).fetch_optional(&mut **tx).await? + else { + return Ok(None); + }; + let execution = DB::string(&activity, "id")?; + let policy = DB::optional_document_row(&activity, "activity_options")? + .and_then(|options| options["cancellation_policy"].as_str().map(str::to_owned)) + .unwrap_or_else(|| "try_cancel".to_owned()); + match policy.as_str() { + "abandon" => { + if matches!(DB::string(&activity, "status")?.as_str(), "pending" | "running") + && DB::optional_instant(&activity, "schedule_to_close_deadline_at")?.is_none() + { + return Err(refuse(StatusCode::CONFLICT, "cancellation_activity_abandon_not_supported")); + } + Ok(None) + } + "wait_cancellation_completed" => { + Self::deliver_activity_cancellation(tx, run_id, request_id, sequence, now()).await?; + Ok((!Self::activity_cancellation_resolved(tx, run_id, request_id, &execution).await?).then_some(execution)) + } + _ => { + Self::deliver_activity_cancellation(tx, run_id, request_id, sequence, now()).await?; + Ok(None) + } + } + } + + /// `ActivityCancellationWait::resume`: after a stop receipt, a whole-run + /// delivery held for its activities wakes with one ready workflow task, + /// once every activity it waits on has resolved. Only while the run still + /// waits on this request, before its cleanup deadline and before any + /// delivery, and when no workflow task is open. Returns whether it woke. + pub(super) async fn resume_activity_cancellation_wait( + tx: &mut Transaction<'_, DB>, + run_id: &str, + request_id: &str, + execution: &str, + receipt_event: &str, + ) -> Result { + let Some(run) = Self::query("SELECT status,queue,cancellation_request_command_id,cancellation_deadline_at,cancellation_delivery_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_optional(&mut **tx).await? + else { + return Ok(false); + }; + if DB::string(&run, "status")? != "waiting" + || DB::optional_string(&run, "cancellation_request_command_id")?.as_deref() != Some(request_id) + || DB::optional_number(&run, "cancellation_delivery_sequence")?.is_some() + || DB::optional_instant(&run, "cancellation_deadline_at")?.is_none_or(|deadline| deadline <= now()) + { + return Ok(false); + } + if Self::scalar("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status IN ('ready','leased')") + .bind(run_id).fetch_one(&mut **tx).await? > 0 + { + return Ok(false); + } + // The latest held delivery for this request names what it waits on. + let mut waiting = None; + for row in Self::query("SELECT payload FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status='completed' ORDER BY created_at DESC,id DESC LIMIT 20") + .bind(run_id).fetch_all(&mut **tx).await? + { + let payload = DB::optional_document_row(&row, "payload")?.unwrap_or(Value::Null); + if payload["cancellation_activity_wait"]["request_id"] == request_id { + waiting = Some(payload["cancellation_activity_wait"]["execution_ids"].clone()); + break; + } + } + let Some(executions) = waiting.as_ref().and_then(Value::as_array) else { + return Ok(false); + }; + if !executions.iter().any(|id| id == execution) { + return Ok(false); + } + for id in executions.iter().filter_map(Value::as_str) { + if !Self::activity_cancellation_resolved(tx, run_id, request_id, id).await? { + return Ok(false); + } + } + let queue = DB::string(&run, "queue")?; + Self::create_task_dispatched(tx, run_id, &queue, "workflow", json!({"resume_source_kind":"activity_cancellation_acknowledged", + "resume_source_id":receipt_event,"workflow_command_id":request_id}), false).await?; + Ok(true) + } + + /// `ActivityCancellationCompletion::resolved` for one execution under + /// the request: it resolved before this request cancelled it, it was + /// cancelled before any attempt started, or the cancelled attempt's + /// worker acknowledged the stop. + pub(super) async fn activity_cancellation_resolved( + tx: &mut Transaction<'_, DB>, + run_id: &str, + request_id: &str, + execution: &str, + ) -> Result { + let mut attempt = None; + let mut cancelled = false; + for row in Self::query("SELECT event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND workflow_command_id=$2 AND event_type IN ('ActivityCancelled','ActivityCancellationAcknowledged')") + .bind(run_id).bind(request_id).fetch_all(&mut **tx).await? + { + let payload = DB::document_row(&row, "payload")?; + if payload["activity_execution_id"] != execution { + continue; + } + match DB::string(&row, "event_type")?.as_str() { + "ActivityCancelled" => { + cancelled = true; + attempt = payload["activity_attempt_id"].as_str().map(str::to_owned); + } + _ => {} + } + } + let Some(attempt) = attempt.filter(|_| cancelled) else { + return Ok(true); + }; + let acknowledged = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND workflow_command_id=$2 AND event_type='ActivityCancellationAcknowledged'") + .bind(run_id).bind(request_id).fetch_all(&mut **tx).await? + .iter() + .filter_map(|row| DB::document_row(row, "payload").ok()) + .any(|payload| payload["activity_execution_id"] == execution && payload["activity_attempt_id"] == attempt.as_str()); + Ok(acknowledged) + } + /// `ActivityCancellation::record` for a cooperative request: the open /// execution at `sequence`, its current attempt and that attempt's own /// task are cancelled, and `ActivityCancelled` names the request. PHP @@ -716,19 +920,20 @@ where } /// `cancelOpenWaitAtSequence` for each member, in sequence order: a - /// pending timer is cancelled, and an open activity is cancelled under - /// the request with the default `try_cancel` policy. + /// pending timer is cancelled, and an open activity is handled under its + /// own cancellation policy. Returns the activities the delivery waits on. async fn cancel_members( tx: &mut Transaction<'_, DB>, run: &DB::Row, request_id: &str, task_id: &str, members: &[(i64, &'static str)], - ) -> Result<()> { + ) -> Result> { let run_id = DB::string(run, "id")?; + let mut waits = Vec::new(); for (member, kind) in members { if *kind == "activity" { - Self::deliver_activity_cancellation(tx, &run_id, request_id, *member, now()).await?; + waits.extend(Self::cancel_activity_member(tx, &run_id, request_id, *member).await?); } else if let Some(timer) = Self::query("SELECT * FROM workflow_run_timers WHERE workflow_run_id=$1 AND sequence=$2") .bind(&run_id).bind(*member).fetch_optional(&mut **tx).await? && DB::string(&timer, "status")? == "pending" @@ -736,7 +941,7 @@ where Self::cancel_root_timer(tx, run, request_id, &timer, Some(task_id), now()).await?; } } - Ok(()) + Ok(waits) } /// `validateCallBoundary` for an activity or timer call. Every typed step diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 3d15633f6..6280943cc 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2957,6 +2957,7 @@ where "schedule_to_start_timeout", "schedule_to_close_timeout", "heartbeat_timeout", + "cancellation_policy", ], )?; text(command, "activity_type")?; @@ -3090,27 +3091,19 @@ where cooperative_child |= super::children::policies(command)?.cooperative(); } if cooperative_child { - let mut unavailable = Vec::new(); - if !protocol - .and_then(|version| version.strip_prefix("1.")) - .and_then(|minor| minor.parse::().ok()) - .is_some_and(|minor| minor >= 20) - { - unavailable.push("request_protocol"); - } - if !Self::claim_supports_cancellation(&task)? { - unavailable.push("worker_claim_capability"); - } - if !unavailable.is_empty() { - return Err(super::RuntimeError::Protocol { - status: StatusCode::CONFLICT, - response: json!({"task_id": task_id, "workflow_task_attempt": body["workflow_task_attempt"], - "worker_id": DB::optional_string(&task, "lease_owner")?, "outcome": "rejected", "recorded": false, - "reason": "child_cancellation_policy_not_supported", "required_capability": "cooperative_cancellation", - "minimum_protocol_version": "1.20", "requested_version": protocol, "unavailable": unavailable, - "remediation": "Use a runtime with child cancellation policies and a claim from a cooperative worker on protocol 1.20 or newer."}), - }); - } + Self::cancellation_policy_guard(&task, task_id, &body, protocol, false, None, "child_cancellation_policy_not_supported", + "Use a runtime with child cancellation policies and a claim from a cooperative worker on protocol 1.20 or newer.")?; + } + // `guardRemoteActivityCancellationPoliciesAvailable`: an explicit + // activity cancellation policy, any value, needs protocol 1.20 and a + // cooperative claim (Server #958). + if commands.iter().any(|command| command["type"] == "schedule_activity" && !command["cancellation_policy"].is_null()) { + // Native has no post-run authority for an abandoned activity yet: + // its task would be stranded once the run closes (Server #971). + let abandon = commands.iter().any(|command| command["type"] == "schedule_activity" && command["cancellation_policy"] == "abandon"); + Self::cancellation_policy_guard(&task, task_id, &body, protocol, true, + abandon.then_some("installed_runtime_activity_policy"), "activity_cancellation_policy_not_supported", + "Use a runtime with remote activity cancellation policies and a cooperative worker claim on protocol 1.20 or newer.")?; } // `guardPortableWorkerAffinityCompletion`, then the normalizer's // local activity rules inside the transaction. @@ -3444,6 +3437,9 @@ where // worker session; its snapshot and the scope's members read it // from there. let mut activity_options = activity_options; + if let Some(policy) = command.get("cancellation_policy").filter(|policy| !policy.is_null()) { + activity_options.get_or_insert_with(|| json!({}))["cancellation_policy"] = policy.clone(); + } if let Some(scope) = command.get("cancellation_scope_id") { activity_options.get_or_insert_with(|| json!({}))["cancellation_scope_id"] = scope.clone(); } @@ -4618,6 +4614,13 @@ where payload["activity"][field] = json!(value); } } + // `ActivitySnapshot::fromExecution`: an abandoned activity names + // the lifetime it keeps after the run stops waiting. + if options["cancellation_policy"] == "abandon" + && let Some(deadline) = DB::optional_instant(activity, "schedule_to_close_deadline_at")? + { + payload["activity"]["schedule_to_close_deadline_at"] = json!(deadline.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()); + } } // Released history readers retain the first claim and heartbeat across // a retry even while the next attempt has not produced new progress. diff --git a/rust/tests/execution/activity_cancellation_ack.rs b/rust/tests/execution/activity_cancellation_ack.rs index 6ce97c951..ee08b82ba 100644 --- a/rust/tests/execution/activity_cancellation_ack.rs +++ b/rust/tests/execution/activity_cancellation_ack.rs @@ -22,11 +22,7 @@ async fn a_retried_activity_acknowledges_on_its_live_task() { 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":"cleanup-worker","task_queue":"test","runtime":"php", - "supported_workflow_types":["echo"],"supported_activity_types":["stop-activity"], - "capabilities":["cooperative_cancellation"]}), "1.20").await; - assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + register_cooperative(&app).await; let started = start(&app, "retried-ack").await; let task = cooperative_poll(&app, "workflow").await; let scheduled = finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"stop-activity", @@ -97,11 +93,7 @@ pub(super) async fn deliver_at(app: &Router, task: &Value, request_id: &Value, s /// timer at sequence 2 is still pending, with the request's workflow task /// claimed. Returns the run, the claimed task and the request id. async fn resolved_activity_then_timer(app: &Router, workflow: &str) -> (Value, Value, Value) { - let registered = request_protocol(app, "POST", "/api/worker/register", - json!({"worker_id":"cleanup-worker","task_queue":"test","runtime":"php", - "supported_workflow_types":["echo"],"supported_activity_types":["stop-activity"], - "capabilities":["cooperative_cancellation"]}), "1.20").await; - assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + register_cooperative(app).await; let started = start(app, workflow).await; let task = cooperative_poll(app, "workflow").await; let scheduled = finish_task(app, &task, json!([{"type":"schedule_activity","activity_type":"stop-activity", @@ -194,3 +186,108 @@ async fn an_activity_delivery_at_the_next_sequence_needs_no_authored_activity() runtime.close().await; database.remove().await; } + +/// Registers the cooperative `cleanup-worker` on protocol 1.20. +async fn register_cooperative(app: &Router) { + let registered = request_protocol(app, "POST", "/api/worker/register", + json!({"worker_id":"cleanup-worker","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":["stop-activity"], + "capabilities":["cooperative_cancellation"]}), "1.20").await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); +} + +/// Starts a run whose first command schedules `stop-activity` with `options`, +/// leases the activity and requests the run's cancellation. Returns the run, +/// the leased activity task and the request id. +async fn cancel_with_policy(app: &Router, workflow: &str, options: Value) -> (Value, Value, Value) { + register_cooperative(app).await; + let started = start(app, workflow).await; + let task = cooperative_poll(app, "workflow").await; + let mut command = json!({"type":"schedule_activity","activity_type":"stop-activity", + "arguments":envelope(Payload::Array(vec![])),"start_to_close_timeout":300}); + for (field, value) in options.as_object().unwrap() { + command[field] = value.clone(); + } + // An explicit activity policy needs protocol 1.20 and a cooperative claim. + let scheduled = request_protocol(app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), + completion(&task, json!([command])), "1.20").await; + assert_eq!(scheduled.0, StatusCode::OK, "{}", scheduled.1); + let activity = cooperative_poll(app, "activity").await; + let requested = request(app, "POST", &format!("/api/workflows/{workflow}/request-cancellation"), + json!({"reason":"stop","cleanup_timeout_seconds":60})).await; + assert_eq!(requested.0, StatusCode::ACCEPTED, "{}", requested.1); + (started, activity, requested.1["cancellation_request"]["request_id"].clone()) +} + +async fn deliver(app: &Router, task: &Value, request_id: &Value) -> (StatusCode, Value) { + request_protocol(app, "POST", + &format!("/api/worker/workflow-tasks/{}/deliver-cancellation", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "request_id":request_id,"sequence":1,"call_kind":"activity"}), "1.20").await +} + +// `wait_cancellation_completed` holds a whole-run delivery until the live +// attempt's worker acknowledges the stop, as PHP Core's +// ActivityCancellationWait does (Server #958). +#[tokio::test] +async fn a_waiting_activity_holds_the_run_delivery_until_its_stop_is_acknowledged() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (started, activity, request_id) = cancel_with_policy(&app, "wait-for-stop", + json!({"cancellation_policy":"wait_cancellation_completed"})).await; + let first = cooperative_poll(&app, "workflow").await; + let (status, held) = deliver(&app, &first, &request_id).await; + assert_eq!(status, StatusCode::OK, "{held}"); + assert_eq!((&held["delivered"], &held["reason"], &held["claim_released"]), + (&json!(false), &json!("cancellation_waiting_for_activity"), &json!(true)), "{held}"); + assert!(task_has_status(&database, &activity["task_id"], "cancelled").await, "the live attempt is cancelled"); + assert!(task_has_status(&database, &first["task_id"], "completed").await, "the claim is released"); + // Nothing resumes the run before the stop receipt. + assert_eq!(cooperative_poll(&app, "workflow").await, Value::Null); + + let acknowledged = acknowledge(&app, &activity["task_id"], &activity["activity_attempt_id"], &activity["lease_owner"], &request_id).await; + assert_eq!((acknowledged.0, acknowledged.1["acknowledged"].clone()), (StatusCode::OK, json!(true)), "{}", acknowledged.1); + let repeated = acknowledge(&app, &activity["task_id"], &activity["activity_attempt_id"], &activity["lease_owner"], &request_id).await; + assert_eq!(repeated.1["duplicate"], true, "{}", repeated.1); + + // One resumed task delivers; a duplicate receipt made no second one. + let resumed = cooperative_poll(&app, "workflow").await; + assert!(resumed.is_object(), "the stop receipt resumes the run"); + let (status, delivered) = deliver(&app, &resumed, &request_id).await; + assert_eq!((status, delivered["delivered"].clone()), (StatusCode::OK, json!(true)), "{delivered}"); + // Payloads read as text on each engine (PostgreSQL stores json, which + // has no LIKE), then counted here. + let run = started["run_id"].as_str().unwrap(); + let payloads = text_rows(&database, + &format!("SELECT id, payload FROM workflow_tasks WHERE workflow_run_id='{run}'"), + &format!("SELECT id::text, payload::text FROM workflow_tasks WHERE workflow_run_id='{run}'"), + &format!("SELECT CAST(id AS CHAR), CAST(payload AS CHAR) FROM workflow_tasks WHERE workflow_run_id='{run}'")).await; + let resumes = payloads.iter().filter(|(_, payload)| payload.contains("activity_cancellation_acknowledged")).count(); + assert_eq!(resumes, 1, "{payloads:?}"); + runtime.close().await; + database.remove().await; +} + +// Native refuses an explicit `abandon` until abandoned activities can +// outlive their run (Server #971), with PHP's refusal for a runtime that +// lacks the policy. An explicit policy on protocol 1.19 is refused too. +#[tokio::test] +async fn an_explicit_abandon_or_a_pre_1_20_policy_is_refused_at_scheduling() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register_cooperative(&app).await; + start(&app, "abandon-refused").await; + let task = cooperative_poll(&app, "workflow").await; + let refused = request_protocol(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), + completion(&task, json!([{"type":"schedule_activity","activity_type":"stop-activity","arguments":envelope(Payload::Array(vec![])), + "start_to_close_timeout":300,"schedule_to_close_timeout":600,"cancellation_policy":"abandon"}])), "1.20").await; + assert_eq!((refused.0, refused.1["reason"].clone(), refused.1["unavailable"].clone()), + (StatusCode::CONFLICT, json!("activity_cancellation_policy_not_supported"), json!(["installed_runtime_activity_policy"])), "{}", refused.1); + let old = finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"stop-activity", + "arguments":envelope(Payload::Array(vec![])),"cancellation_policy":"try_cancel"}])).await; + assert_eq!((old.0, old.1["unavailable"].clone()), (StatusCode::CONFLICT, json!(["request_protocol"])), "{}", old.1); + runtime.close().await; + database.remove().await; +}