Rust #325: serve workflow-task cancellation scopes (open, prepare, deliver)

.github/workflows/server-parity.yml

@@ -45,6 +45,7 @@ on:
       - tests/Unit/ServerParityRecordLocalActivityContractTest.mjs
       - tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs
       - tests/Unit/ServerParityActivityCancellationAckContractTest.mjs
+      - tests/Unit/ServerParityCancellationScopesContractTest.mjs
       - tests/Unit/ServerParityServiceCatalogContractTest.mjs
       - tests/Unit/ServerParityWorkflowImportContractTest.mjs
       - tests/Unit/ServerParityWaterlineImportContractTest.mjs
@@ -1803,7 +1804,7 @@ jobs:
           bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \
             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/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.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 tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.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 tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.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

docs/rust-server-port.md

@@ -82,12 +82,12 @@ Updated October 24, 2026.
     ([#737](https://github.com/durable-workflow/server/issues/737)).
 
   No expectation is edited to make native pass.
-- **Parity surface.** Native serves 140 of the 175 API route methods in
+- **Parity surface.** Native serves 143 of the 175 API route methods in
   published 2.11.1's route table, with message streams, update validation,
-  activity cancellation acknowledgement, worker sessions, and embedded v2
-  and Waterline v1 workflow import. An execution test sends each one through
-  the native router, and the ones still answering
-  `501 rust_capability_not_implemented` must match
+  activity cancellation acknowledgement, worker sessions, workflow-task
+  cancellation scopes, and embedded v2 and Waterline v1 workflow import. An
+  execution test sends each one through the native router, and the ones
+  still answering `501 rust_capability_not_implemented` must match
   [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt).
   Unlike published PHP, native's embedded v2 import refuses a bundle whose
   rows reference rows outside it, links only runs of one namespace, and reads
@@ -139,7 +139,8 @@ any other unknown path or method answers PHP's own 404 or 405
 - **Worker protocol features.**
   - Local activities (prepare, checkpoint, recover, heartbeat, control,
     outcome and cancellation acknowledgement).
-  - Cancellation scopes.
+  - Cancellation scope checkpoints. (Open, prepare and deliver are served,
+    with the `cancellation-scopes` fixture.)
   - (Worker sessions are served: create, heartbeat, close, list and detail,
     with the `worker-sessions` and `worker-session-activities` fixtures.)
   - (Activity cancellation acknowledgement is served:

rust/src/runtime/activity_cancellation_ack.rs

@@ -103,7 +103,12 @@ store_methods! {
         {
             return Err(refused("activity_cancellation_acknowledgement_fence_mismatch"));
         }
-        if DB::optional_string(run, "cancellation_request_command_id")?.as_deref() != Some(request_id) {
+        // The run's request, or a scope request the run recorded.
+        let (_, facts) = Self::scope_run(&mut tx, &run_id).await?;
+        let events = Self::scope_events(&mut tx, &run_id).await?;
+        if facts.request_id.as_deref() != Some(request_id)
+            && !events.iter().any(|event| event.kind == "CancellationScopeRequested" && event.payload["request_id"] == request_id)
+        {
             return Err(refused("cancellation_request_mismatch"));
         }
         let Some((cancellation_event, cancellation)) =
@@ -117,7 +122,14 @@ store_methods! {
         {
             return Err(refused("activity_cancellation_snapshot_mismatch"));
         }
-        let Some(context) = Self::cooperative_context(&mut tx, &run_id, request_id).await? else {
+        // `ActivityCancellationContext::forEvent`: a scoped fence's accepted
+        // scope context, or the run request's.
+        let Some(context) = events.iter().find(|event| event.id == cancellation_event)
+            .and_then(|event| super::cancellation_scopes::cancellation_facts(&events, &facts, event))
+            .filter(|(request, _, _)| request == request_id)
+            .and_then(|(_, root_request_id, deadline)| Some(CooperativeContext { root_request_id,
+                deadline: chrono::DateTime::parse_from_rfc3339(&deadline).ok()?.with_timezone(&chrono::Utc) }))
+        else {
             return Err(refused("cancellation_context_not_recorded"));
         };
         if let Some((event, _)) = Self::acknowledgement_event(&mut tx, &run_id, &cancellation_event, attempt_id).await? {
@@ -156,20 +168,21 @@ store_methods! {
         execution_id: &str,
         attempt_id: &str,
     ) -> Result<Option<Value>> {
-        let Some(request_id) = DB::optional_string(run, "cancellation_request_command_id")? else {
-            return Ok(None);
-        };
+        // The execution's original fence, whichever request recorded it.
         let run_id = DB::string(run, "id")?;
-        let Some((cancellation_event, _)) =
-            Self::activity_cancellation_event(tx, &run_id, &request_id, execution_id, attempt_id).await?
+        let (_, facts) = Self::scope_run(tx, &run_id).await?;
+        let events = Self::scope_events(tx, &run_id).await?;
+        let Some(cancelled) = events.iter().find(|event| event.kind == "ActivityCancelled" && event.payload["activity_execution_id"] == execution_id)
+            .filter(|event| event.payload["activity_attempt_id"] == attempt_id)
         else {
             return Ok(None);
         };
-        let Some(context) = Self::cooperative_context(tx, &run_id, &request_id).await? else {
+        let Some((request_id, root_request_id, deadline)) = super::cancellation_scopes::cancellation_facts(&events, &facts, cancelled) else {
             return Ok(None);
         };
-        let mut receipt = json!({"request_id":request_id,"root_request_id":context.root_request_id,
-            "cleanup_deadline_at":micros(context.deadline),"cancellation_history_event_id":cancellation_event,
+        let cancellation_event = cancelled.id.clone();
+        let mut receipt = json!({"request_id":request_id,"root_request_id":root_request_id,
+            "cleanup_deadline_at":deadline,"cancellation_history_event_id":cancellation_event,
             "callback_state":"unknown","history_event_id":null,"acknowledged_at":null,"received_after_deadline":null});
         if let Some((event, payload)) = Self::acknowledgement_event(tx, &run_id, &cancellation_event, attempt_id).await? {
             receipt["callback_state"] = payload["callback_state"].clone();
@@ -216,26 +229,6 @@ store_methods! {
         }
         Ok(None)
     }
-
-    /// The request's root and original cleanup deadline, from its
-    /// `CooperativeCancellationRequested` event.
-    async fn cooperative_context(tx: &mut Transaction<'_, DB>, run_id: &str, request_id: &str) -> Result<Option<CooperativeContext>> {
-        let Some(requested) = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND workflow_command_id=$2 AND event_type='CooperativeCancellationRequested' ORDER BY sequence LIMIT 1")
-            .bind(run_id).bind(request_id).fetch_optional(&mut **tx).await?
-        else {
-            return Ok(None);
-        };
-        let cancellation = &DB::document_row(&requested, "payload")?["cancellation"];
-        let Some(deadline) = cancellation["cleanup_deadline_at"].as_str()
-            .and_then(|value| chrono::DateTime::parse_from_rfc3339(value).ok())
-            .map(|value| value.with_timezone(&chrono::Utc))
-        else {
-            return Ok(None);
-        };
-        let root_request_id = cancellation["lineage"].as_array().and_then(|lineage| lineage.first())
-            .and_then(|root| root["request_id"].as_str()).unwrap_or(request_id).to_owned();
-        Ok(Some(CooperativeContext { root_request_id, deadline }))
-    }
 }
 
 struct CooperativeContext {