Rust #325: deliver a parent's cooperative cancellation to a retried child

.github/workflows/server-parity.yml

@@ -68,6 +68,7 @@ on:
       - tests/Unit/ServerParityChildDescriptionsContractTest.mjs
       - tests/Unit/ServerParityChildRetriesContractTest.mjs
       - tests/Unit/ServerParityChildVisibilityContractTest.mjs
+      - tests/Unit/ServerParityRetriedChildCancellationContractTest.mjs
       - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs
       - tests/Unit/ServerParityQueryArgumentsContractTest.mjs
       - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs
@@ -2171,6 +2172,7 @@ jobs:
               tests/Unit/ServerParityChildDescriptionsContractTest.mjs \
               tests/Unit/ServerParityChildRetriesContractTest.mjs \
               tests/Unit/ServerParityChildVisibilityContractTest.mjs \
+              tests/Unit/ServerParityRetriedChildCancellationContractTest.mjs \
               tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs \
               tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs \
               tests/Unit/ServerParityManyUpdatesContractTest.mjs \

docs/rust-server-port-checkpoints/2026-10-29-retried-child-cancellation.md

@@ -0,0 +1,33 @@
+# October 29, cooperative cancellation of a retried child (#873)
+
+- **Before.** After #869, a retried child call has a link per attempt, and
+  the call names the latest attempt's run. Native's cooperative cancellation
+  expected one link per call:
+  - `child_link_policies` rejected the call's links, so a parent's
+    cancellation request was refused (`422
+    cooperative_cancellation_propagation_not_available`);
+  - the delivery boundary also required one link and one `ChildRunStarted`
+    naming the call's cancellation policy.
+
+  Security's review of #869 found the gap.
+- **Contract** (published Server 2.11.7). The request is admitted and
+  delivered at the child boundary, and the parent completes as cancelled.
+  - Under `abandon`, the retry keeps running.
+  - Under `try_cancel`, the parent records `ChildCancellationRequested`
+    naming the retry run, and that run, not the failed first run, records
+    `CooperativeCancellationRequested`.
+  - PHP's retry `ChildRunStarted` carries no `cancellation_policy`.
+- **Fixture.** `retried-child-cancellation`, raw worker HTTP on protocol
+  1.20 in a fresh namespace, with a cooperative parent worker and a separate
+  child worker. One case per policy (`abandon`, `try_cancel`). It records
+  the answers, the parent's child and cancellation events, the child's runs
+  and the retry run's events.
+- **Native.**
+  - `child_link_policies` accepts a retried call's links: the first under
+    the call's id, each retry under its own, and earlier attempts.
+  - The delivery boundary reads every link of the call and picks the one
+    for the call's current run. It accepts the call's scheduled event, and
+    one started event per attempt; a retry's may omit the cancellation
+    policy.
+- **Tests.** `a_parents_cooperative_cancellation_reaches_a_retried_child`
+  (`child_retries.rs`) covers both policies.

docs/rust-server-port.md

@@ -90,7 +90,6 @@ Updated October 27, 2026.
   `501 rust_capability_not_implemented` must match
   [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt).
 - **Remaining for the release candidate:**
-  - landing the 2.11.5 reference move;
   - the scope question with the operator: prepared local activities,
     workflow-bound Nexus handlers and external payload takeover;
   - publishing the native image (General runs the release);
@@ -614,9 +613,12 @@ contract fixture before its correction.
   ([checkpoint](rust-server-port-checkpoints/2026-10-29-child-retries.md)).
   A child timeout or backoff too far away for a timestamp is refused
   (`invalid_child_timeout`, `invalid_child_retry_policy`), as a timer delay
-  is. Remaining: native's cooperative cancellation delivery expects one
-  link per call, so it refuses one delivered to a retried child
-  (`cancellation_delivery_shape_mismatch`; #873, pending a PHP recording).
+  is. A parent's cooperative cancellation reaches a retried child (#873):
+  the request is admitted, and the delivery follows the call to its current
+  run and link, so under `try_cancel` the retry run, not the failed first
+  run, is asked to cancel. The `retried-child-cancellation` fixture records
+  both policies against published 2.11.7
+  ([checkpoint](rust-server-port-checkpoints/2026-10-29-retried-child-cancellation.md)).
 - **Trace context (Core 2.10, published in Server 2.11.1).** Core adds
   nullable `observability_context` columns on runs, commands and tasks.
   Native has no carrier: it accepts and ignores a top-level carrier on
@@ -686,9 +688,9 @@ contract fixture before its correction.
   and PHP's message, and records nothing.
 - **Cooperative cancellation boundaries.** Published PHP delivers a run's
   cancellation at an activity, local activity, timer, condition, signal,
-  child, parallel or selection-handle boundary. Native delivers it at a
-  timer, a condition or signal wait, a child and a parallel group whose
-  members are all timers. It refuses the other boundaries, including
+  child, parallel or selection-handle boundary. Native delivers it at an
+  activity, a timer, a condition or signal wait, a child (retried or not)
+  and a parallel group whose members are all timers. It refuses the other boundaries, including
   parallel groups with other member kinds, with
   `422 cancellation_delivery_boundary_not_available`.
 - **Activity results after a deadline, before its timeout runs.** Published

rust/src/runtime/children.rs

@@ -197,14 +197,19 @@ where
         };
         let metadata = DB::document_row(call, "metadata")?;
         let cancellation = metadata["cancellation_policy"].as_str().unwrap_or("abandon").to_owned();
+        // A retried call (`markChildCallRetryStarted`) has a link per attempt:
+        // the first under the call's id, each retry under its own, and the
+        // call names the latest attempt's run.
+        let retried = metadata["attempt_count"].as_i64().is_some_and(|attempts| attempts > 1);
         let consistent = DB::string(call, "parent_workflow_instance_id")?
             == DB::string(link, "parent_workflow_instance_id")?
             && DB::optional_string(call, "resolved_child_instance_id")?
                 == Some(DB::string(link, "child_workflow_instance_id")?)
-            && DB::optional_string(call, "resolved_child_run_id")?
-                == Some(DB::string(link, "child_workflow_run_id")?)
+            && (retried
+                || DB::optional_string(call, "resolved_child_run_id")?
+                    == Some(DB::string(link, "child_workflow_run_id")?))
             && DB::optional_string(call, "parent_close_policy")?.as_deref() == Some(parent_close.as_str())
-            && metadata["child_call_id"] == DB::string(link, "id")?
+            && (retried || metadata["child_call_id"] == DB::string(link, "id")?)
             && matches!(
                 DB::string(call, "status")?.as_str(),
                 "started" | "completed" | "failed" | "cancelled"

rust/src/runtime/cooperative.rs

@@ -710,27 +710,47 @@
         if found { Ok(()) } else { Err(mismatch()) }
     }
 
     async fn validate_child_cancellation_boundary(
         tx: &mut Transaction<'_, DB>,
         run_id: &str,
         sequence: i64,
         request_sequence: i64,
     ) -> Result<super::child_cancellation::ChildBoundary> {
         let mismatch = || refuse(StatusCode::CONFLICT, "cancellation_delivery_shape_mismatch");
+        // One link per attempt of the call: a retried child
+        // (`startChildRetryIfAvailable`) adds a link per retry, and the call
+        // names the current attempt's run.
         let links = Self::query(
-            "SELECT * FROM workflow_links WHERE parent_workflow_run_id=$1 AND sequence=$2 LIMIT 2",
+            "SELECT * FROM workflow_links WHERE parent_workflow_run_id=$1 AND sequence=$2 ORDER BY id LIMIT 101",
         )
         .bind(run_id)
         .bind(sequence)
         .fetch_all(&mut **tx)
         .await?;
-        let link = links
-            .first()
-            .filter(|_| links.len() == 1)
-            .ok_or_else(mismatch)?;
-        let (parent_close, cancellation) = Self::child_link_policies(tx, link).await?.ok_or_else(mismatch)?;
-        let call = Self::query("SELECT child_workflow_type,child_workflow_class FROM workflow_child_calls WHERE parent_workflow_run_id=$1 AND sequence=$2")
+        if links.is_empty() || links.len() > 100 {
+            return Err(mismatch());
+        }
+        let call = Self::query("SELECT child_workflow_type,child_workflow_class,resolved_child_run_id,metadata FROM workflow_child_calls WHERE parent_workflow_run_id=$1 AND sequence=$2")
             .bind(run_id).bind(sequence).fetch_one(&mut **tx).await?;
+        let metadata = DB::document_row(&call, "metadata")?;
+        let call_id = metadata["child_call_id"].as_str().map(str::to_owned).ok_or_else(mismatch)?;
+        let retried = metadata["attempt_count"].as_i64().is_some_and(|attempts| attempts > 1);
+        if links.len() > 1 && !retried {
+            return Err(mismatch());
+        }
+        let current_run = DB::optional_string(&call, "resolved_child_run_id")?.ok_or_else(mismatch)?;
+        let mut current = None;
+        let mut first = None;
+        for link in &links {
+            if DB::string(link, "child_workflow_run_id")? == current_run {
+                current = Some(link);
+            }
+            if DB::string(link, "id")? == call_id {
+                first = Some(link);
+            }
+        }
+        let (current, first) = (current.ok_or_else(mismatch)?, first.ok_or_else(mismatch)?);
+        let (parent_close, cancellation) = Self::child_link_policies(tx, current).await?.ok_or_else(mismatch)?;
         let events = Self::query("SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ChildWorkflowScheduled','ChildRunStarted','ChildRunCompleted','ChildRunFailed','ChildRunCancelled','ChildRunTerminated') ORDER BY sequence LIMIT 1001")
             .bind(run_id).fetch_all(&mut **tx).await?;
         if events.len() > 1000 {
@@ -740,7 +760,7 @@
             ));
         }
         let mut scheduled = false;
-        let mut started = false;
+        let mut started_links: Vec<String> = Vec::new();
         let mut resolved_before_request = false;
         for event in events {
             let payload = DB::document_row(&event, "payload")?;
@@ -749,16 +769,35 @@
             }
             let kind = DB::string(&event, "event_type")?;
             if matches!(kind.as_str(), "ChildWorkflowScheduled" | "ChildRunStarted") {
-                if payload["workflow_link_id"] != DB::string(link, "id")?
-                    || payload["child_call_id"] != DB::string(link, "id")?
+                // The call's first event names the first link; each started
+                // event names its own attempt's link and run.
+                let link = if kind == "ChildWorkflowScheduled" {
+                    first
+                } else {
+                    let named = payload["workflow_link_id"].as_str().unwrap_or_default();
+                    let mut found = None;
+                    for link in &links {
+                        if DB::string(link, "id")? == named {
+                            found = Some(link);
+                        }
+                    }
+                    found.ok_or_else(mismatch)?
+                };
+                let link_id = DB::string(link, "id")?;
+                // PHP records a retry's `ChildRunStarted` without the call's
+                // cancellation policy.
+                let retry = kind == "ChildRunStarted" && payload["retry_attempt"].is_i64();
+                if payload["workflow_link_id"] != link_id.as_str()
+                    || payload["child_call_id"] != call_id.as_str()
                     || payload["child_workflow_instance_id"]
                         != DB::string(link, "child_workflow_instance_id")?
                     || payload["child_workflow_run_id"]
                         != DB::string(link, "child_workflow_run_id")?
                     || payload["child_workflow_type"] != DB::string(&call, "child_workflow_type")?
                     || payload["child_workflow_class"] != DB::string(&call, "child_workflow_class")?
                     || payload["parent_close_policy"] != parent_close.as_str()
-                    || payload["cancellation_policy"] != cancellation.as_str()
+                    || (payload["cancellation_policy"] != cancellation.as_str()
+                        && !(retry && payload.get("cancellation_policy").is_none()))
                     || payload.get("child_workflow").is_some()
                 {
                     return Err(mismatch());
@@ -772,20 +811,22 @@
                         "cancellation_delivery_requires_scope",
                     ));
                 }
-                let seen = if kind == "ChildWorkflowScheduled" {
-                    &mut scheduled
+                if kind == "ChildWorkflowScheduled" {
+                    if scheduled {
+                        return Err(mismatch());
+                    }
+                    scheduled = true;
                 } else {
-                    &mut started
-                };
-                if *seen {
-                    return Err(mismatch());
+                    if started_links.contains(&link_id) {
+                        return Err(mismatch());
+                    }
+                    started_links.push(link_id);
                 }
-                *seen = true;
             } else if DB::number(&event, "sequence")? < request_sequence {
                 resolved_before_request = true;
             }
         }
-        if !scheduled || !started {
+        if !scheduled || !started_links.contains(&DB::string(current, "id")?) {
             return Err(mismatch());
         }
         if resolved_before_request {
@@ -795,8 +836,8 @@
             ));
         }
         Ok(super::child_cancellation::ChildBoundary {
-            child_instance: DB::string(link, "child_workflow_instance_id")?,
-            child_run: DB::string(link, "child_workflow_run_id")?,
+            child_instance: DB::string(current, "child_workflow_instance_id")?,
+            child_run: DB::string(current, "child_workflow_run_id")?,
             policy: cancellation,
         })
     }

rust/tests/execution/child_retries.rs

@@ -207,3 +207,57 @@ async fn a_child_and_its_retry_inherit_the_parents_memos_and_search_attributes()
     runtime.close().await;
     database.remove().await;
 }
+
+// #873: published PHP admits and delivers a parent's cooperative cancellation
+// to a child it retried. The delivery follows the call to its current run.
+#[tokio::test]
+async fn a_parents_cooperative_cancellation_reaches_a_retried_child() {
+    for policy in ["abandon", "try_cancel"] {
+        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":[],
+            "capabilities":["cooperative_cancellation"]}), "1.20").await;
+        assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1);
+        register(&app, "child-worker", json!(["child"]), json!([])).await;
+        let parent = start(&app, "cooperative-retried").await;
+        let task = cooperative_poll(&app, "workflow").await;
+        // A child cancellation policy needs a cooperative claim on protocol 1.20.
+        let (status, body) = request_protocol(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete",
+            task["task_id"].as_str().unwrap()), completion(&task, json!([child(json!({"retry_policy":{"max_attempts":2},
+            "cancellation_policy":policy}))])), "1.20").await;
+        assert_eq!(status, StatusCode::OK, "{policy}: {body}");
+        let first = poll(&app, "child-worker", "workflow").await;
+        let (status, body) = finish_task(&app, &first, fail(false)).await;
+        assert_eq!(status, StatusCode::OK, "{policy}: {body}");
+        let requested = request(&app, "POST", "/api/workflows/cooperative-retried/request-cancellation",
+            json!({"reason":"stop the retried child","cleanup_timeout_seconds":30})).await;
+        assert_eq!(requested.0, StatusCode::ACCEPTED, "{policy}: {}", requested.1);
+        let pending = requested.1["cancellation_request"].clone();
+        let root = cooperative_poll(&app, "workflow").await;
+        let delivered = request_protocol(&app, "POST", &format!("/api/worker/workflow-tasks/{}/deliver-cancellation",
+            root["task_id"].as_str().unwrap()), json!({"lease_owner":root["lease_owner"],
+            "workflow_task_attempt":root["workflow_task_attempt"],"request_id":pending["request_id"],"sequence":1,
+            "call_kind":"child"}), "1.20").await;
+        assert_eq!(delivered.0, StatusCode::OK, "{policy}: {}", delivered.1);
+        let (status, body) = finish_task(&app, &root, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await;
+        assert_eq!((status, &body["run_status"]), (StatusCode::OK, &json!("cancelled")), "{policy}: {body}");
+        let events = run_history(&app, &parent["workflow_id"], &parent["run_id"]).await;
+        let kinds: Vec<_> = events.as_array().unwrap().iter().map(|event| event["event_type"].as_str().unwrap().to_owned())
+            .filter(|kind| kind.starts_with("Child") || kind.contains("Cancel")).collect();
+        let mut expected = vec!["ChildWorkflowScheduled", "ChildRunStarted", "ChildRunStarted", "CooperativeCancellationRequested"];
+        if policy == "try_cancel" {
+            expected.push("ChildCancellationRequested");
+        }
+        expected.extend(["CooperativeCancellationDelivered", "WorkflowCancelled"]);
+        assert_eq!(kinds, expected, "{policy}");
+        // The retry run, not the failed first run, takes a requested child cancellation.
+        let retry = child_events(&events)[2]["payload"].clone();
+        let retry_history = run_history(&app, &retry["child_workflow_instance_id"], &retry["child_workflow_run_id"]).await;
+        let requested_child = retry_history.as_array().unwrap().iter().any(|event| event["event_type"] == "CooperativeCancellationRequested");
+        assert_eq!(requested_child, policy == "try_cancel", "{policy}: {retry_history}");
+        runtime.close().await;
+        database.remove().await;
+    }
+}