Implement PHP Server's sticky execution contract on native (#1017 part 1)

rust/src/runtime/local_activities.rs

@@ -934,10 +934,14 @@ pub(super) async fn exceptions(records: &BTreeMap<usize, Record>, codec: Arc<Sem
 
 /// `portableWorkerAffinityRejection`.
 fn affinity_rejection(task_id: &str, attempt: &Value, reason: &str, requested: Option<&str>) -> RuntimeError {
+    capability_rejection(task_id, attempt, reason, "local_activities", requested)
+}
+
+fn capability_rejection(task_id: &str, attempt: &Value, reason: &str, capability: &str, requested: Option<&str>) -> RuntimeError {
     RuntimeError::Protocol {
         status: StatusCode::CONFLICT,
         response: json!({"task_id": task_id, "workflow_task_attempt": attempt, "outcome": "rejected", "recorded": false,
-            "reason": reason, "capability": "local_activities", "requested_version": requested,
+            "reason": reason, "capability": capability, "requested_version": requested,
             "minimum_protocol_version": MINIMUM_PROTOCOL_VERSION,
             "remediation": "Use cold durable replay unless the registered worker manifest explicitly supports the required capability and exact cache identity.",
             "protocol_version": super::WORKER_PROTOCOL_VERSION, "server_capabilities": super::store::capabilities()}),
@@ -973,6 +977,67 @@ store_methods! {
     }
 
 
+    /// `guardPortableWorkerAffinityCompletion` for a completion that claims
+    /// sticky execution (Server #1017): protocol 1.18 or newer, a lease owner
+    /// whose registration lists and supports `sticky_execution`, and a claim
+    /// naming exactly that worker, the task's run and workflow, and the
+    /// registration's build (else its SDK version).
+    pub(super) async fn guard_sticky_cache(tx: &mut Transaction<'_, DB>, namespace: &str, task: &DB::Row, body: &Value,
+        protocol: Option<&str>, workflow_id: &str) -> Result<()> {
+        const CAPABILITY: &str = "sticky_execution";
+        let task_id = DB::string(task, "id")?;
+        let attempt = body["workflow_task_attempt"].as_i64().map_or(Value::Null, Value::from);
+        let rejected = |reason| capability_rejection(&task_id, &attempt, reason, CAPABILITY, protocol);
+        if below(protocol, MINIMUM_PROTOCOL_VERSION) {
+            return Err(rejected("portable_worker_affinity_protocol_floor_unavailable"));
+        }
+        let owner = super::text(body, "lease_owner")?;
+        let worker = Self::query("SELECT capabilities,capability_manifest,build_id,sdk_version FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2")
+            .bind(namespace).bind(owner).fetch_optional(&mut **tx).await?;
+        let Some(worker) = worker else {
+            return Err(rejected("worker_capability_not_supported"));
+        };
+        let listed = DB::optional_document_row(&worker, "capabilities")?.as_ref().and_then(Value::as_array)
+            .is_some_and(|capabilities| capabilities.iter().any(|capability| capability.as_str() == Some(CAPABILITY)));
+        let supported = DB::optional_document_row(&worker, "capability_manifest")?.as_ref().and_then(|manifest| manifest.get(CAPABILITY))
+            .is_some_and(|entry| (entry.is_object() || entry.is_array()) && entry["supported"] == Value::Bool(true));
+        if !listed || !supported {
+            return Err(rejected("worker_capability_not_supported"));
+        }
+        let claim = &body["sticky_cache"];
+        let build = DB::optional_string(&worker, "build_id")?.or(DB::optional_string(&worker, "sdk_version")?);
+        let text = |field: &str| claim[field].as_str().unwrap_or_default();
+        if text("worker_id") != owner || text("run_id") != DB::string(task, "workflow_run_id")? || text("workflow_id") != workflow_id
+            || build.as_deref().is_none_or(|build| text("build_id") != build)
+        {
+            return Err(rejected("sticky_cache_identity_mismatch"));
+        }
+        Ok(())
+    }
+
+    /// `applyStickyCacheClaim`: after the accepted completion, in its
+    /// transaction, the run keeps the worker's affinity until the claim's
+    /// expiry, or loses it when it closed. The worker's reported cache
+    /// metrics are kept with its registration.
+    pub(super) async fn record_sticky_cache(tx: &mut Transaction<'_, DB>, namespace: &str, run_id: &str, body: &Value) -> Result<()> {
+        let claim = &body["sticky_cache"];
+        let owner = super::text(body, "lease_owner")?;
+        let status = Self::query("SELECT status FROM workflow_runs WHERE id=$1").bind(run_id).fetch_one(&mut **tx).await?;
+        let retain = !super::sticky_execution::terminal(&DB::string(&status, "status")?);
+        let until = super::store::now() + chrono::Duration::seconds(claim["ttl_seconds"].as_i64().unwrap_or(0));
+        Self::query("UPDATE workflow_runs SET sticky_worker_id=$1,sticky_until=$2 WHERE id=$3")
+            .bind(retain.then_some(owner)).bind(DB::optional_time(retain.then_some(until))).bind(run_id)
+            .execute(&mut **tx).await?;
+        let mut metrics = Self::query("SELECT process_metrics FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2")
+            .bind(namespace).bind(owner).fetch_optional(&mut **tx).await?
+            .map(|row| DB::optional_document_row(&row, "process_metrics")).transpose()?.flatten()
+            .filter(Value::is_object).unwrap_or_else(|| json!({}));
+        metrics["sticky_cache"] = claim.get("metrics").filter(|value| !value.is_null()).cloned().unwrap_or_else(|| json!([]));
+        Self::query("UPDATE workflow_worker_registrations SET process_metrics=$1 WHERE namespace=$2 AND worker_id=$3")
+            .bind(DB::document(&metrics)).bind(namespace).bind(owner).execute(&mut **tx).await?;
+        Ok(())
+    }
+
     /// `local_activity_timestamp_out_of_range`, in the completion's response.
     fn timestamp_out_of_range(tx: &Transaction<'_, DB>) -> RuntimeError {
         const REASON: &str = "local_activity_timestamp_out_of_range";

rust/src/runtime/mod.rs

@@ -195,6 +195,7 @@ mod terminal_updates;
 pub mod timer_holds;
 mod timer_summary;
 mod transaction;
+mod sticky_execution;
 mod update_contracts;
 mod update_summary;
 mod update_validation;

rust/src/runtime/operator_metrics.rs

@@ -275,7 +275,43 @@
             "update_wait": {"completion_timeout_seconds": 10, "poll_interval_milliseconds": 50},
             "repair_policy": repair_policy(),
             "matching_role": matching_role(config),
-            "sticky_execution": sticky_execution(),
+            "sticky_execution": self.sticky_execution_metrics(now).await?,
+        }))
+    }
+
+    /// `OperatorMetrics::stickyExecution` for this namespace (Server #1017):
+    /// live affinity, and the replay each workflow claim of the last minute
+    /// told its worker to expect.
+    async fn sticky_execution_metrics(&self, now: DateTime<Utc>) -> Result<Value> {
+        let namespace = self.namespace.as_ref();
+        let active = Self::scalar("SELECT COUNT(*) FROM workflow_runs WHERE namespace=$1 AND status IN ('pending','running','waiting') AND sticky_worker_id IS NOT NULL AND sticky_worker_id<>'' AND sticky_until>$2")
+            .bind(namespace).bind(DB::bind_time(now)).fetch_one(&self.pool).await?;
+        let ready = Self::scalar("SELECT COUNT(*) FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.task_type='workflow' AND t.status='ready' AND r.sticky_worker_id IS NOT NULL AND r.sticky_worker_id<>'' AND r.sticky_until>$2")
+            .bind(namespace).bind(DB::bind_time(now)).fetch_one(&self.pool).await?;
+        let leased = Self::scalar("SELECT COUNT(*) FROM workflow_tasks WHERE namespace=$1 AND task_type='workflow' AND status='leased' AND sticky_worker_id IS NOT NULL")
+            .bind(namespace).fetch_one(&self.pool).await?;
+        let mut counts = std::collections::BTreeMap::new();
+        for row in Self::query("SELECT sticky_replay_mode AS mode,COUNT(*) AS claims FROM workflow_tasks WHERE namespace=$1 AND sticky_claimed_at>=$2 AND sticky_replay_mode IS NOT NULL GROUP BY sticky_replay_mode")
+            .bind(namespace).bind(DB::bind_time(now - chrono::Duration::seconds(60))).fetch_all(&self.pool).await?
+        {
+            counts.insert(DB::string(&row, "mode")?, DB::number(&row, "claims")?);
+        }
+        let count = |mode: &str| counts.get(mode).copied().unwrap_or(0);
+        let (hits, forced) = (count(super::sticky_execution::STICKY_HIT_EXPECTED), count(super::sticky_execution::FORCED_COLD_REPLAY));
+        let rate = |part: i64| if hits + forced == 0 { 0.0 } else { part as f64 / (hits + forced) as f64 };
+        Ok(json!({
+            "enabled": true,
+            "ttl_seconds": 300,
+            "active_sticky_runs": active,
+            "ready_sticky_tasks": ready,
+            "leased_sticky_tasks": leased,
+            "hit_expected_last_minute": hits,
+            "miss_last_minute": forced,
+            "forced_cold_replay_last_minute": forced,
+            "cold_replay_last_minute": count(super::sticky_execution::COLD_REPLAY),
+            "hit_rate_last_minute": rate(hits),
+            "miss_rate_last_minute": rate(forced),
+            "capacity_pressure_tasks": ready,
         }))
     }
 
@@ -1800,23 +1836,7 @@
     })
 }
 
 /// Native routes no task to a sticky worker.
-fn sticky_execution() -> Value {
-    json!({
-        "enabled": false,
-        "ttl_seconds": 0,
-        "active_sticky_runs": 0,
-        "ready_sticky_tasks": 0,
-        "leased_sticky_tasks": 0,
-        "hit_expected_last_minute": 0,
-        "miss_last_minute": 0,
-        "forced_cold_replay_last_minute": 0,
-        "cold_replay_last_minute": 0,
-        "hit_rate_last_minute": 0.0,
-        "miss_rate_last_minute": 0.0,
-        "capacity_pressure_tasks": 0,
-    })
-}
 
 #[cfg(test)]
 mod tests {

rust/src/runtime/sticky_execution.rs

@@ -0,0 +1,79 @@
+//! Sticky workflow execution as published PHP Server 2.11.8 and Core
+//! 2.15.12 run it (Server #1017).
+//!
+//! A worker's workflow task completion may claim the run with `sticky_cache`.
+//! The run then records that worker and the claim's expiry. Routing does not
+//! change: any eligible worker may claim the run's next task, and every claim
+//! still delivers full history. The claim only tells the worker which replay
+//! to expect: `sticky_hit_expected` when the sticky worker claims within the
+//! expiry, `forced_cold_replay` when the affinity expired or another worker
+//! claimed, and `cold_replay` when the run has no affinity.
+use chrono::{DateTime, Utc};
+use serde_json::{Value, json};
+
+pub(super) const STICKY_HIT_EXPECTED: &str = "sticky_hit_expected";
+pub(super) const COLD_REPLAY: &str = "cold_replay";
+pub(super) const FORCED_COLD_REPLAY: &str = "forced_cold_replay";
+const DEFAULT_TTL_SECONDS: i64 = 300;
+
+/// `StickyExecution::describe()`, the server capability PHP advertises.
+pub(super) fn describe() -> Value {
+    json!({
+        "feature": "sticky_execution",
+        "supported": true,
+        "worker_capability": "sticky_execution",
+        "minimum_worker_protocol_version": super::local_activities::MINIMUM_PROTOCOL_VERSION,
+        "kind": "v2_replay_optimization",
+        "correctness_fallback": "cold_replay",
+        "routing_identity": "worker_id",
+        "cache_owner": "worker_process",
+        "cache_key_fields": ["workflow_id", "run_id", "build_id"],
+        "durable_affinity_fields": ["sticky_worker_id", "sticky_until"],
+        "task_diagnostic_fields": ["sticky_replay_mode", "sticky_claimed_at"],
+        "replay_modes": [STICKY_HIT_EXPECTED, COLD_REPLAY, FORCED_COLD_REPLAY],
+        "metrics": ["hit", "miss", "eviction", "forced_cold_replay"],
+        "default_ttl_seconds": DEFAULT_TTL_SECONDS,
+        "default_cache_capacity": 100,
+        "default_capacity_pressure_ratio": 0.9,
+        "contract": [
+            "sticky caches are process-local and owned by the worker that advertises them",
+            "sticky affinity expires and ordinary cold replay is always the correctness fallback",
+            "worker replacement, restart, drain, or cache eviction may force cold replay",
+            "workflow code must remain replay-safe and must not rely on process-local state for correctness",
+        ],
+    })
+}
+
+/// `StickyExecution::claimReplayMode` for a claim by `lease_owner` of a run
+/// whose affinity is `worker` until `until`.
+pub(super) fn claim_mode(worker: Option<&str>, until: Option<DateTime<Utc>>, lease_owner: &str, now: DateTime<Utc>) -> &'static str {
+    let Some(worker) = worker.filter(|worker| !worker.is_empty()) else {
+        return COLD_REPLAY;
+    };
+    if until.is_none_or(|until| until <= now) {
+        return FORCED_COLD_REPLAY;
+    }
+    if worker == lease_owner { STICKY_HIT_EXPECTED } else { FORCED_COLD_REPLAY }
+}
+
+/// A run in one of these states keeps no affinity.
+pub(super) fn terminal(status: &str) -> bool {
+    matches!(status, "completed" | "failed" | "cancelled" | "terminated" | "continued")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn claims_follow_cores_replay_rule() {
+        let now = Utc::now();
+        let later = Some(now + chrono::Duration::seconds(60));
+        assert_eq!(claim_mode(None, later, "a", now), COLD_REPLAY);
+        assert_eq!(claim_mode(Some(""), later, "a", now), COLD_REPLAY);
+        assert_eq!(claim_mode(Some("a"), later, "a", now), STICKY_HIT_EXPECTED);
+        assert_eq!(claim_mode(Some("a"), later, "b", now), FORCED_COLD_REPLAY);
+        assert_eq!(claim_mode(Some("a"), Some(now), "a", now), FORCED_COLD_REPLAY);
+        assert_eq!(claim_mode(Some("a"), None, "a", now), FORCED_COLD_REPLAY);
+    }
+}

rust/src/runtime/store.rs

@@ -1960,6 +1960,10 @@ where
             }
         };
         if previous.is_some() {
+            // `releaseLeasedWorkflowTasksForReplacedWorker`: a replaced worker
+            // process holds no sticky cache, so its runs lose their affinity.
+            Self::query("UPDATE workflow_runs SET sticky_worker_id=NULL,sticky_until=NULL WHERE namespace=$1 AND sticky_worker_id=$2")
+                .bind(self.namespace.as_ref()).bind(worker_id).execute(&mut *tx).await?;
             // Match PHP's registration boundary: return original workflow leases
             // to ready without a repair command/history event. Token rotation
             // and lease release commit together; failure restores the old attempt.
@@ -2689,9 +2693,20 @@ where
                     "updated_at":DB::instant(&worker,"updated_at")?,"last_heartbeat_at":DB::instant(&worker,"last_heartbeat_at")?,
                     "task_id":task_id,"run_id":run_id,"lease_owner":worker_id,"attempt":attempt});
             }
-            Self::query("UPDATE workflow_tasks SET ready_origin_at=CASE WHEN status='leased' AND task_type='workflow' THEN $1 ELSE ready_origin_at END,status='leased',lease_owner=$2,lease_expires_at=$3,attempt_count=$4,leased_at=$5,payload=$6 WHERE id=$7")
+            // `claimStatus`: a workflow claim records the replay its worker
+            // should expect from the run's sticky affinity (Server #1017).
+            let (sticky_worker, sticky_until, sticky_mode) = if kind == "workflow" {
+                let worker = DB::optional_string(&run, "sticky_worker_id")?.filter(|worker| !worker.is_empty());
+                let until = DB::optional_instant(&run, "sticky_until")?;
+                let mode = super::sticky_execution::claim_mode(worker.as_deref(), until, worker_id, leased_at);
+                (worker, until.filter(|_| mode != super::sticky_execution::COLD_REPLAY), Some(mode))
+            } else {
+                (None, None, None)
+            };
+            Self::query("UPDATE workflow_tasks SET ready_origin_at=CASE WHEN status='leased' AND task_type='workflow' THEN $1 ELSE ready_origin_at END,status='leased',lease_owner=$2,lease_expires_at=$3,attempt_count=$4,leased_at=$5,payload=$6,sticky_worker_id=$7,sticky_until=$8,sticky_replay_mode=$9,sticky_claimed_at=$10 WHERE id=$11")
                 .bind(DB::optional_time(ready_origin)).bind(worker_id).bind(DB::bind_time(expires)).bind(attempt).bind(DB::bind_time(leased_at))
-                .bind(DB::document(&task_payload)).bind(&task_id).execute(&mut **tx).await?;
+                .bind(DB::document(&task_payload)).bind(sticky_worker.as_deref()).bind(DB::optional_time(sticky_until)).bind(sticky_mode)
+                .bind(DB::optional_time(sticky_mode.map(|_| leased_at))).bind(&task_id).execute(&mut **tx).await?;
             // A lease is worker authority, not an authored lifecycle commit.
             if kind == "workflow" {
                 Self::query("UPDATE workflow_run_waits SET task_status='leased',updated_at=$1 WHERE workflow_run_id=$2 AND task_id=$3 AND ((kind='child' AND status IN ('resolved','cancelled')) OR (kind='update' AND status='open'))")
@@ -2770,7 +2785,9 @@ where
                 if let Some(update_id) = task_payload["workflow_update_id"].as_str() {
                     claim["workflow_update_id"] = json!(update_id);
                 }
-                claim["sticky_replay_mode"] = json!("cold_replay");
+                claim["sticky_worker_id"] = json!(sticky_worker);
+                claim["sticky_until"] = json!(sticky_until.map(super::cooperative::timestamp));
+                claim["sticky_replay_mode"] = json!(sticky_mode);
                 claim["compatibility"] = json!(DB::optional_string(&task, "compatibility")?
                     .filter(|build| !build.trim().is_empty())
                     .or(DB::optional_string(&run, "compatibility")?.filter(|build| !build.trim().is_empty())));
@@ -3416,6 +3433,10 @@ where
             local = super::local_activities::normalize(commands, protocol)?;
             exceptions = super::local_activities::exceptions(&local, transition_codec.clone()).await?;
         }
+        let sticky = body.get("sticky_cache").is_some_and(|claim| !claim.is_null());
+        if sticky {
+            Self::guard_sticky_cache(&mut tx, self.namespace.as_ref(), &task, &body, protocol, &stream_instance).await?;
+        }
         if let Some(checkpoint) = checkpoint {
             // Server #994: Core's refused checkpoint for a run that closed
             // after the guard is the same not-recorded receipt.
@@ -3695,6 +3716,9 @@ where
                 None => super::carried_polls::Outcome::Declined,
             }
         };
+        if sticky {
+            Self::record_sticky_cache(&mut tx, self.namespace.as_ref(), &run_id, &body).await?;
+        }
         let transaction = if deferred { Some(tx) } else {
             tx.commit().await?;
             self.wake.notify_waiters();
@@ -5789,7 +5813,7 @@ fn capabilities_template() -> Value {
         "workflow_memo_updates": false, "cooperative_cancellation": true, "prepared_local_activities": false,
         "prepared_local_activity_groups": false, "prepared_local_activity_cancellation_policies": [],
         "worker_sessions": super::worker_sessions::capability(), "worker_session_verbs": ["create", "heartbeat", "close"],
-        "sticky_execution": false, "local_activities": super::local_activities::manifest(), "message_streams": super::message_streams::worker_capability(),
+        "sticky_execution": super::sticky_execution::describe(), "local_activities": super::local_activities::manifest(), "message_streams": super::message_streams::worker_capability(),
         "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(),
         "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
         "claimed_task_ready_age": true, "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,

rust/tests/execution.rs

@@ -87,6 +87,8 @@ mod activity_failure_fields;
 mod query_worker_results;
 #[path = "execution/rejected_updates.rs"]
 mod rejected_updates;
+#[path = "execution/sticky_execution.rs"]
+mod sticky_execution;
 #[path = "execution/early_update_contracts.rs"]
 mod early_update_contracts;
 #[path = "execution/continue_messages.rs"]