Rust #325: record local activities like Server 2.11.1

.github/workflows/server-parity.yml

@@ -42,6 +42,7 @@ on:
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityOperatorMetricsContractTest.mjs
       - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs
+      - tests/Unit/ServerParityRecordLocalActivityContractTest.mjs
       - tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs
       - tests/Unit/ServerParityActivityCancellationAckContractTest.mjs
       - tests/Unit/ServerParityServiceCatalogContractTest.mjs
@@ -1802,7 +1803,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/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/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

@@ -232,6 +232,33 @@ contract fixture before its correction.
   batch size, and still caps parallel groups at 100 members and a run at 64
   updates over its lifetime. #612 tracks the rest
   ([checkpoint](rust-server-port-checkpoints/2026-10-23-structural-limits.md)).
+- **Completion budget, ahead of published PHP.** Native applies Phase A of
+  [Workflow #930](https://github.com/durable-workflow/workflow/issues/930)
+  ([Workflow #934](https://github.com/durable-workflow/workflow/pull/934)),
+  which published 2.11.1 does not have yet:
+  - `record_local_activity` caps `duration_ms` and heartbeat `elapsed_ms` at
+    7 days and `backoff_seconds` at 1 day, each with PHP's field error. The walk
+    back from completion is overflow-checked, and it is refused with
+    `local_activity_timestamp_out_of_range` before it reaches
+    1970-01-01T00:00:01Z.
+  - After a run's deadline close, which is never metered, a completion whose
+    local activities' exact events and rows exceed
+    `DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE` or `_ROWS` (other commands count one
+    event) is refused with `completion_budget_exceeded`.
+  - The rest of the completion runs under a meter. Each history event charges
+    one event, its bytes as `HistoryBudget` measures them (against
+    `_BYTES`) and two rows, and each local activity row is charged before it
+    is written. A refusal answers `history_transaction_limit_exceeded`, writes
+    nothing and leaves the task's lease as it was. A tripped meter also refuses
+    the commit.
+  - Native never stores an external payload object while applying a
+    completion: workers upload objects first, and a completion only claims
+    their references. Native also dispatches no lifecycle events. So #934's
+    staging and event deferral have no native counterpart.
+  - Operator metrics, the health snapshot and the backend capabilities
+    publish the active `history_transaction_size`, `history_transaction_bytes`
+    and `history_transaction_rows` (0 when disabled), as Workflow #934's
+    `StructuralLimits::snapshot()` does.
 - **Waterline v1 projections, ahead of published PHP.**
   - Native refuses a message stream append to a projection with
     `409 v1_projection_read_only`, at the route before the body is read and
@@ -245,14 +272,10 @@ contract fixture before its correction.
     event is built; then each event is built and measured in turn, and the
     first event past the byte limit stops the build. Published 2.11.1 has no limit
     ([#770](https://github.com/durable-workflow/server/issues/770)).
-  - Operator metrics, the health snapshot and the backend capabilities
-    publish the active `history_transaction_size` and
-    `history_transaction_bytes` (0 when disabled), as Workflow #934's
-    `StructuralLimits::snapshot()` does.
 - **Environment settings.** Of PHP's 164 contract settings, native honors
   49 and ignores 51 that change nothing it serves. It also honors
-  `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES`, which the Workflow package reads
-  and PHP's contract does not list yet. It refuses to start on a
+  `DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES` and `_ROWS`, which the Workflow
+  package reads and PHP's contract does not list yet. It refuses to start on a
   non-default value of the other 64
   ([checkpoint](rust-server-port-checkpoints/2026-10-22-environment-contract.md)).
   Several of those are cheap to honor: repair timing and the multi-node

rust/src/config_audit.rs

@@ -548,6 +548,7 @@ const UNLISTED: &[Setting] = &[
 /// their legacy names warn as the contract's do.
 pub const NATIVE_HONORED: &[Setting] = &[
     Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"), disposition: Honored },
+    Setting { name: "DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS", legacy: Some("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_ROWS"), disposition: Honored },
 ];
 
 /// Every `vars` entry of `config/dw-contract.php`, with its contract legacy name.
@@ -1086,7 +1087,7 @@ mod tests {
     fn the_history_transaction_limits_are_honored_settings() {
         let honored = |name: &str| SETTINGS.iter().chain(NATIVE_HONORED).find(|setting| setting.name == name)
             .is_some_and(|setting| matches!(setting.disposition, Honored));
-        for name in ["DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES"] {
+        for name in ["DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS"] {
             assert!(honored(name), "{name}");
         }
         // NATIVE_HONORED names are outside PHP's contract and listed once.
@@ -1095,15 +1096,17 @@ mod tests {
             assert!(!SETTINGS.iter().any(|listed| listed.name == setting.name));
         }
         // Strict mode starts with each primary name set.
-        let mut primary = environment(&[("DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "123"), ("DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567")]);
+        let mut primary = environment(&[("DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "123"), ("DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567"),
+            ("DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS", "89")]);
         primary.insert("DW_ENV_AUDIT_STRICT".into(), "1".into());
         let report = audit(&primary);
         assert_eq!((report.warnings.as_slice(), report.refusal()), (&[] as &[String], None));
         // A legacy name warns as deprecated, naming its primary, and nothing else.
         let report = audit(&environment(&[("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE", "123"),
-            ("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567")]));
+            ("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES", "4567"), ("WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_ROWS", "89")]));
         assert_eq!(report.warnings, [
             "Legacy env var WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_BYTES is still honored but deprecated — rename to DW_V2_LIMIT_HISTORY_TRANSACTION_BYTES.",
+            "Legacy env var WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_ROWS is still honored but deprecated — rename to DW_V2_LIMIT_HISTORY_TRANSACTION_ROWS.",
             "Legacy env var WORKFLOW_V2_LIMIT_HISTORY_TRANSACTION_SIZE is still honored but deprecated — rename to DW_V2_LIMIT_HISTORY_TRANSACTION_SIZE.",
         ]);
         assert_eq!(report.refusal(), None);

rust/src/runtime/activity_metadata.rs

@@ -498,12 +498,15 @@ fn present_attempt(attempt: &Value, execution: Option<&Value>) -> Value {
     });
     let cancelled =
         status == Some("cancelled") || execution.is_some_and(|row| row["status"] == "cancelled");
+    // Workflow 2.10's attempt view keeps a recorded lease loss as the
+    // expired attempt's failure reason.
+    let failure = string(&attempt["failure_reason"]);
     let reason = if status == Some("cancelled") {
         Some("attempt_cancelled")
+    } else if status == Some("expired") {
+        Some(failure.unwrap_or(if cancelled { "activity_cancelled" } else { "attempt_expired" }))
     } else if cancelled {
         Some("activity_cancelled")
-    } else if status == Some("expired") {
-        Some("attempt_expired")
     } else if matches!(status, Some("completed" | "failed")) {
         Some("attempt_closed")
     } else if execution
@@ -518,7 +521,7 @@ fn present_attempt(attempt: &Value, execution: Option<&Value>) -> Value {
     let mut result = json!({"id":attempt["id"],"worker_attempt_id":attempt["worker_attempt_id"],
         "attempt_number":attempt["attempt_number"],"status":attempt["status"],"task_id":attempt["workflow_task_id"],
         "lease_owner":attempt["lease_owner"],"last_heartbeat_progress":progress(&attempt["last_heartbeat_progress"]),
-        "can_continue":can_continue,"cancel_requested":cancelled,"stop_reason":reason});
+        "can_continue":can_continue,"cancel_requested":cancelled,"stop_reason":reason,"failure_reason":failure});
     for field in [
         "started_at",
         "last_heartbeat_at",
@@ -559,7 +562,7 @@ fn synthetic_attempt(state: &Value, row: &Value) -> Option<Value> {
     };
     let mut result = json!({"id":id,"attempt_number":count,"status":status,"task_id":null,
         "lease_owner":null,"lease_expires_at":null,"can_continue":can_continue,
-        "cancel_requested":cancelled,"stop_reason":reason,
+        "cancel_requested":cancelled,"stop_reason":reason,"failure_reason":null,
         "last_heartbeat_progress":progress(&state["last_heartbeat_progress"])});
     for field in ["started_at", "last_heartbeat_at", "closed_at"] {
         result[field] = first(&[&state[field], &row[field]]);
@@ -806,15 +809,22 @@ mod tests {
         assert_eq!(cases.len(), 32);
         for case in cases {
             let input = &case["input"];
-            assert_eq!(
-                project(
-                    input["events"].as_array().unwrap(),
-                    input["executions"].as_array().unwrap()
-                ),
-                case["expected"],
-                "{}",
-                case["name"]
+            let mut projected = project(
+                input["events"].as_array().unwrap(),
+                input["executions"].as_array().unwrap(),
             );
+            // Published Workflow 2.11.1's RunActivityView adds each attempt's
+            // `failure_reason` (a recorded lease loss, otherwise null). This
+            // corpus comes from 2.5.22, before that field, and none of its
+            // cases records a lease loss, so the field must be null here and
+            // the rest is compared as recorded.
+            for activity in projected.as_array_mut().into_iter().flatten() {
+                for attempt in activity["attempts"].as_array_mut().into_iter().flatten() {
+                    let reason = attempt.as_object_mut().unwrap().remove("failure_reason");
+                    assert_eq!(reason, Some(Value::Null), "{}", case["name"]);
+                }
+            }
+            assert_eq!(projected, case["expected"], "{}", case["name"]);
         }
     }
 }

rust/src/runtime/activity_summary.rs

@@ -13,6 +13,8 @@ use std::collections::BTreeMap;
 #[derive(Default)]
 pub(super) struct ActivitySummaryHistory {
     open: BTreeMap<String, ExpectedActivity>,
+    /// Recorded local activities not yet closed by their terminal event.
+    local: BTreeMap<String, (i64, String)>,
 }
 
 struct ExpectedActivity {
@@ -50,15 +52,12 @@ impl ActivitySummaryHistory {
         {
             return false;
         }
-        // Local execution still requires its separately qualified history.
-        if snapshot["id"] != id
-            || payload["sequence"] != sequence
-            || payload["activity_type"] != kind
-            || snapshot["local_activity"] == true
-            || snapshot["execution_mode"] == "local"
-        {
+        if snapshot["id"] != id || payload["sequence"] != sequence || payload["activity_type"] != kind {
             return false;
         }
+        if snapshot["local_activity"] == true || snapshot["execution_mode"] == "local" {
+            return self.apply_local(event, id, sequence, kind, attempts, payload);
+        }
         let status = match event {
             "ActivityScheduled" | "ActivityRetryScheduled" => "pending",
             "ActivityStarted" | "ActivityHeartbeatRecorded" => "running",
@@ -127,6 +126,38 @@ impl ActivitySummaryHistory {
         }
         true
     }
+
+    /// A `record_local_activity` writes its whole history in the completion
+    /// that reports it: scheduled, then each attempt, then the terminal event,
+    /// which closes it without a wait. A local activity left open (a prepared
+    /// one) is not qualified here.
+    fn apply_local(&mut self, event: &str, id: &str, sequence: i64, kind: &str, attempts: i64, payload: &Value) -> bool {
+        let status = match event {
+            "ActivityScheduled" | "ActivityRetryScheduled" => "pending",
+            "ActivityStarted" | "ActivityHeartbeatRecorded" => "running",
+            "ActivityCompleted" => "completed",
+            "ActivityFailed" | "ActivityTimedOut" => "failed",
+            "ActivityCancelled" => "cancelled",
+            _ => return false,
+        };
+        if payload["execution_mode"] != "local" || payload["local_activity"] != true || payload["activity"]["status"] != status {
+            return false;
+        }
+        if event == "ActivityScheduled" {
+            if attempts != 0 || self.open.contains_key(id) || self.local.contains_key(id) {
+                return false;
+            }
+            self.local.insert(id.to_owned(), (sequence, kind.to_owned()));
+            return true;
+        }
+        if self.local.get(id) != Some(&(sequence, kind.to_owned())) {
+            return false;
+        }
+        if matches!(status, "completed" | "failed" | "cancelled") {
+            self.local.remove(id);
+        }
+        true
+    }
 }
 
 impl ExpectedActivity {
@@ -184,7 +215,7 @@ where
     ) -> Result<bool> {
         let rows = Self::query("SELECT w.*,a.id AS activity_id,a.sequence AS activity_sequence,a.activity_type,a.status AS activity_status,a.attempt_count,a.current_attempt_id,a.parallel_group_path AS activity_group_path,q.id AS matched_task_id,q.task_type AS matched_task_type,q.payload AS matched_task_payload FROM workflow_run_waits w LEFT JOIN activity_executions a ON a.id=w.resume_source_id AND a.workflow_run_id=w.workflow_run_id LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND w.kind='activity' AND w.status='open' ORDER BY w.sequence,w.opened_at,w.resume_source_id LIMIT 101")
             .bind(run_id).fetch_all(&mut **tx).await?;
-        if rows.len() != history.open.len() || rows.len() > 100 {
+        if rows.len() != history.open.len() || rows.len() > 100 || !history.local.is_empty() {
             return Ok(false);
         }
         let mut selected_activity_task = false;