Rust #325: write only moved wait positions; answer prepared local-activity routes as PHP refuses them

docs/rust-server-port.md

@@ -82,14 +82,15 @@ 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 148 of the 175 API route methods in
+- **Parity surface.** Native answers 156 of the 175 API route methods in
   published 2.11.1's route table, with message streams, durable workflow
   stream writes, update validation, activity cancellation acknowledgement,
   worker sessions, workflow-task cancellation scopes, caller-indexed Nexus
   operation reads, bridge adapter webhooks, 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
+  workflow import. Eight of those are the prepared local-activity routes,
+  which it answers with PHP's refusal for a bridge without the prepared role.
+  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,8 +140,20 @@ any other unknown path or method answers PHP's own 404 or 405
   `repair-pass` fixture; a recovery native does not perform yet is
   reported as its candidate's failure, with 207.
 - **Worker protocol features.**
-  - Local activities (prepare, checkpoint, recover, heartbeat, control,
-    outcome and cancellation acknowledgement).
+  - Prepared local activities (prepare, checkpoint, group checkpoint,
+    recover, heartbeat, control, outcome and cancellation acknowledgement)
+    are answered as a PHP server whose bridge lacks the prepared role answers
+    them. Native advertises `prepared_local_activities: false`, no groups and
+    no policies, so SDKs record local activities with `record_local_activity`.
+    A worker that calls one anyway gets PHP's 409
+    `prepared_local_activity[_groups]_not_supported`, listing
+    `request_protocol` below 1.20 and `installed_runtime_prepared_local_activity`
+    (plus `…_groups` for the group checkpoint), before any task is read.
+    No recorded fixture: published 2.11.1 has the role, so its answers can't
+    be recorded. `prepared_local_routes_are_refused_without_the_prepared_role`
+    pins the answers to `PreparedLocalActivityController::refused`. Serving
+    them is planned in
+    [#325 comment 6423707058](https://github.com/durable-workflow/server/issues/325#issuecomment-6423707058).
   - Cancellation scope checkpoints. (Open, prepare and deliver are served,
     with the `cancellation-scopes` fixture.)
   - (Worker sessions are served: create, heartbeat, close, list and detail,

rust/src/runtime/backend.rs

@@ -19,7 +19,9 @@ pub(super) trait Backend: Database {
     // deferred read transaction already keeps one WAL snapshot.
     const READ_SNAPSHOT_BEGIN: &str = "BEGIN";
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT sequence,event_type,CASE WHEN projection_bytes<=8388608 THEN payload ELSE NULL END AS payload,workflow_task_id,recorded_at,projection_count FROM (SELECT sequence,event_type,payload,workflow_task_id,recorded_at,COUNT(*) OVER () AS projection_count,SUM(octet_length(CAST(payload AS TEXT))) OVER (ORDER BY sequence) AS projection_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') ORDER BY sequence LIMIT 10001) AS page ORDER BY sequence";
-    const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id LIMIT 100 OFFSET $2";
+    // Waits whose stored position differs from their chronological ordinal
+    // over every retained wait of the run, as Workflow #941's projector stores.
+    const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1), ordinals AS (SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE position IS DISTINCT FROM ordinal ORDER BY ordinal LIMIT 100";
     const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS TEXT) AS payload_json FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16";
     const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND payload->>'failure_id'=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
     const OBSERVATION_ATTRIBUTES: &str = "SELECT \"key\" AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND \"key\" IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY \"key\" LIMIT 20";
@@ -171,20 +173,23 @@ impl Backend for Sqlite {
     // whole UTC seconds with SQLite's clock functions and round integer
     // microseconds explicitly; SQLite's fractional-date rounding differs at ties.
     // Preserve Carbon's signed U.u decimal key for pre-epoch dates as well.
-    const OPEN_WAIT_ORDER: &str = r#"
+    const WAIT_POSITION_MOVES: &str = r#"
 WITH utc_waits AS (
-    SELECT id,wait_id,opened_at,
+    SELECT id,wait_id,position,opened_at,
         replace(replace(replace(opened_at,'T',' '),'Z',''),'+00:00','') AS utc_opened
-    FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open'
+    FROM workflow_run_waits WHERE workflow_run_id=$1
 ), clocks AS (
     SELECT *,unixepoch(substr(utc_opened,1,19))*1000000
         +(CASE WHEN unixepoch(substr(utc_opened,1,19))<0 THEN -1 ELSE 1 END)
         *CAST(CASE WHEN substr(utc_opened,20,1)='.'
             THEN substr(substr(utc_opened,21)||'000000',1,6) ELSE '000000' END AS INTEGER) AS micros
     FROM utc_waits
+), ordinals AS (
+    SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,
+        CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id)-1 AS ordinal
+    FROM clocks
 )
-SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,
-    CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id LIMIT 100 OFFSET $2
+SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal LIMIT 100
 "#;
     const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND json_extract(payload,'$.failure_id') COLLATE BINARY=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
     const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (json_extract(payload,'$.activity_execution_id') COLLATE BINARY=$2 OR json_extract(payload,'$.activity.id') COLLATE BINARY=$3) ORDER BY sequence DESC LIMIT 1";
@@ -479,7 +484,7 @@ impl Backend for MySql {
     const READ_SNAPSHOT_BEGIN: &str = "START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY";
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT h.sequence,h.event_type,CASE WHEN page.projection_bytes<=8388608 THEN h.payload ELSE NULL END AS payload,h.workflow_task_id,h.recorded_at,page.projection_count FROM (SELECT id,sequence,COUNT(*) OVER () AS projection_count,SUM(payload_bytes) OVER (ORDER BY sequence) AS projection_bytes FROM (SELECT id,sequence,octet_length(CAST(payload AS CHAR CHARACTER SET utf8mb4)) AS payload_bytes FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled')) AS sizes ORDER BY sequence LIMIT 10001) AS page JOIN workflow_history_events h ON h.id=page.id ORDER BY page.sequence";
     const PHP_CONNECTION: &str = "mysql";
-    const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=? AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id LIMIT 100 OFFSET ?";
+    const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=?), ordinals AS (SELECT id,position,CAST(ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id) AS SIGNED)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE NOT (position <=> ordinal) ORDER BY ordinal LIMIT 100";
     // The run-and-sequence index keeps MySQL from sorting whole rows: a
     // filesort of one large payload exceeds the default sort buffer and
     // fails the transaction ("Out of sort memory").

rust/src/runtime/http.rs

@@ -281,6 +281,19 @@ pub fn router(runtime: Runtime) -> Router {
             "/api/worker/workflow-tasks/{task_id}/cancellation-scopes/deliver",
             post(deliver_cancellation_scope),
         )
+        // Native lacks the prepared local-activity role: refused as PHP
+        // refuses them without it (prepared_local.rs).
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/prepare", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/recover", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint-group", post(prepared_local_group_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/control", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/heartbeat", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/outcome", post(prepared_local_refused))
+        .route(
+            "/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/acknowledge-cancellation",
+            post(prepared_local_refused),
+        )
         .route(
             "/api/worker/workflow-tasks/{task_id}/complete",
             post(complete_workflow),
@@ -817,6 +830,8 @@ fn worker_route(rest: &str) -> Option<Allowed> {
             | ["update-validation-tasks", _, "approve" | "reject"]
             | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat" | "deliver-cancellation"]
             | ["workflow-tasks", _, "cancellation-scopes", "open" | "prepare" | "deliver"]
+            | ["workflow-tasks", _, "local-activities", "prepare" | "recover" | "checkpoint" | "checkpoint-group"]
+            | ["workflow-tasks", _, "local-activities", _, "control" | "heartbeat" | "outcome" | "acknowledge-cancellation"]
             | ["activity-tasks", _, "complete" | "fail" | "heartbeat" | "status" | "renew-lease" | "acknowledge-cancellation"]
     )
     .then_some(Allowed::Any)
@@ -2332,6 +2347,22 @@ async fn deliver_cancellation_scope(
     Ok((status, Json(body)))
 }
 
+/// POST …/local-activities/*: refused before any task is read, as a PHP
+/// server whose bridge lacks the prepared role refuses them.
+async fn prepared_local_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json<Value>)> {
+    prepared_local(&headers, &body, super::prepared_local::Capability::Prepared)
+}
+
+async fn prepared_local_group_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json<Value>)> {
+    prepared_local(&headers, &body, super::prepared_local::Capability::Groups)
+}
+
+fn prepared_local(headers: &HeaderMap, body: &[u8], capability: super::prepared_local::Capability) -> Result<(StatusCode, Json<Value>)> {
+    let protocol = headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok());
+    let (status, response) = super::prepared_local::refusal(&request_json(body)?, protocol, capability);
+    Ok((status, Json(response)))
+}
+
 async fn acknowledge_activity_cancellation(
     Extension(runtime): Extension<Runtime>,
     Path(task_id): Path<String>,

rust/src/runtime/mod.rs

@@ -66,6 +66,7 @@ mod history_presentation;
 mod http;
 mod laravel_validation;
 mod local_activities;
+mod prepared_local;
 mod memo_json;
 mod memo_limits;
 mod history_meter;

rust/src/runtime/prepared_local.rs

@@ -0,0 +1,115 @@
+//! The prepared local-activity worker routes, as a published PHP server
+//! answers them when its installed Workflow bridge lacks the prepared role
+//! (`PreparedLocalActivityController::unavailable` and `refused`, and
+//! `WorkerController::checkpointLocalActivityPrefix` and
+//! `checkpointLocalActivityGroup`).
+//!
+//! Native advertises `prepared_local_activities: false`, no groups and no
+//! cancellation policies, so SDKs record local activities with
+//! `record_local_activity` instead. A worker that calls one of these routes
+//! anyway is refused before any task is read, with PHP's 409 body:
+//! - `request_protocol` when its protocol version is below 1.20;
+//! - `installed_runtime_prepared_local_activity`, always;
+//! - for `checkpoint-group`, also `installed_runtime_prepared_local_activity_groups`.
+//!
+//! A version header native does not support at all is refused earlier, by the
+//! worker protocol gate, with 400 as in PHP. PHP's own server-protocol reason
+//! never applies: native serves cooperative cancellation.
+//!
+//! PHP reports `(string) $request->input('lease_owner', '')` as `worker_id`. An
+//! array or object there is an array-to-string conversion, which Laravel turns
+//! into a 500. Native reports an empty `worker_id` instead.
+use super::laravel_rules;
+use axum::http::StatusCode;
+use serde_json::{Value, json};
+
+/// `PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION`.
+const MINIMUM_PROTOCOL_VERSION: &str = "1.20";
+
+/// Which prepared capability the refused route needs.
+#[derive(Clone, Copy)]
+pub(crate) enum Capability {
+    /// Prepare, recover, control, heartbeat, outcome, acknowledge-cancellation
+    /// and the local prefix checkpoint.
+    Prepared,
+    /// The group checkpoint.
+    Groups,
+}
+
+/// PHP's 409 for a route whose prepared role the installed runtime lacks.
+pub(crate) fn refusal(body: &Value, protocol: Option<&str>, capability: Capability) -> (StatusCode, Value) {
+    let mut unavailable = Vec::new();
+    if !super::worker_admission::at_least(protocol, MINIMUM_PROTOCOL_VERSION) {
+        unavailable.push("request_protocol");
+    }
+    unavailable.push("installed_runtime_prepared_local_activity");
+    let (reason, required) = match capability {
+        Capability::Prepared => ("prepared_local_activity_not_supported", "prepared_local_activities"),
+        Capability::Groups => {
+            unavailable.push("installed_runtime_prepared_local_activity_groups");
+            ("prepared_local_activity_groups_not_supported", "prepared_local_activity_groups")
+        }
+    };
+    let body = laravel_rules::normalize(body);
+    let worker_id = match &body["lease_owner"] {
+        Value::String(text) => text.clone(),
+        Value::Number(number) => php_number(number),
+        Value::Bool(true) => "1".into(),
+        _ => String::new(),
+    };
+    let response = json!({
+        "reason": reason,
+        "required_capability": required,
+        "minimum_protocol_version": MINIMUM_PROTOCOL_VERSION,
+        "worker_id": worker_id,
+        "unavailable": unavailable,
+        "remediation": "Use a prepared-local backend and a claim issued to a capable protocol 1.20 worker.",
+    });
+    (StatusCode::CONFLICT, super::cancellation_scopes::enveloped(response))
+}
+
+/// PHP's `(string)` of a JSON number.
+fn php_number(number: &serde_json::Number) -> String {
+    match number.as_i64() {
+        Some(whole) => whole.to_string(),
+        None => {
+            let float = number.as_f64().unwrap_or_default();
+            if float.fract() == 0.0 && float.abs() < 1e15 { format!("{float:.0}") } else { float.to_string() }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_current_worker_is_refused_for_the_missing_runtime_role_only() {
+        let (status, body) = refusal(&json!({"lease_owner": " worker-1 "}), Some("1.20"), Capability::Prepared);
+        assert_eq!(status, StatusCode::CONFLICT);
+        assert_eq!(body["reason"], "prepared_local_activity_not_supported");
+        assert_eq!(body["required_capability"], "prepared_local_activities");
+        assert_eq!(body["worker_id"], "worker-1");
+        assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity"]));
+    }
+
+    #[test]
+    fn an_older_worker_is_also_refused_for_its_protocol() {
+        let (_, body) = refusal(&json!({}), Some("1.19"), Capability::Groups);
+        assert_eq!(body["reason"], "prepared_local_activity_groups_not_supported");
+        assert_eq!(body["required_capability"], "prepared_local_activity_groups");
+        assert_eq!(body["worker_id"], "");
+        assert_eq!(
+            body["unavailable"],
+            json!(["request_protocol", "installed_runtime_prepared_local_activity", "installed_runtime_prepared_local_activity_groups"])
+        );
+    }
+
+    #[test]
+    fn worker_id_is_php_string_cast_of_lease_owner() {
+        for (owner, expected) in [(json!(7), "7"), (json!(true), "1"), (json!(false), ""), (json!(null), ""), (json!(["x"]), "")] {
+            let (_, body) = refusal(&json!({"lease_owner": owner}), Some("1.20"), Capability::Prepared);
+            assert_eq!(body["worker_id"], expected, "{owner}");
+        }
+    }
+}

rust/src/runtime/transaction.rs

@@ -96,7 +96,7 @@ where
             return Err(refusal);
         }
         for run_id in std::mem::take(&mut self.summaries) {
-            Store::<DB>::order_open_waits(&mut self, &run_id).await?;
+            Store::<DB>::order_waits(&mut self, &run_id).await?;
             Store::<DB>::project_run_summary(&mut self, &run_id).await?;
         }
         self.inner.commit().await?;