diff --git a/rust/src/runtime/local_activities.rs b/rust/src/runtime/local_activities.rs index aa72b7c8e..0179138dd 100644 --- a/rust/src/runtime/local_activities.rs +++ b/rust/src/runtime/local_activities.rs @@ -934,10 +934,14 @@ pub(super) async fn exceptions(records: &BTreeMap, codec: Arc) -> 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"; diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index f9ddd7930..2e47d61c6 100644 --- a/rust/src/runtime/mod.rs +++ b/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; diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index cc16ad0aa..37f11c6e9 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/rust/src/runtime/operator_metrics.rs @@ -275,7 +275,43 @@ where "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) -> Result { + 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, })) } @@ -1801,22 +1837,6 @@ fn matching_role(config: &MetricsConfig) -> Value { } /// 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 { diff --git a/rust/src/runtime/sticky_execution.rs b/rust/src/runtime/sticky_execution.rs new file mode 100644 index 000000000..862dd08dd --- /dev/null +++ b/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>, lease_owner: &str, now: DateTime) -> &'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); + } +} diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 3a45f6646..455f77b5f 100644 --- a/rust/src/runtime/store.rs +++ b/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, diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 1873f5db8..af4696332 100644 --- a/rust/tests/execution.rs +++ b/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"] diff --git a/rust/tests/execution/continuations.rs b/rust/tests/execution/continuations.rs index 6514e202d..a3042a6c4 100644 --- a/rust/tests/execution/continuations.rs +++ b/rust/tests/execution/continuations.rs @@ -337,7 +337,11 @@ async fn continuation_commits_one_successor_and_preserves_original_receipts_afte let database = TestDatabase::new().await; let runtime = database.open().await.unwrap(); let app = router(runtime.clone()); - register(&app, "continuation-worker", json!(["echo"]), json!([])).await; + // A sticky claim needs a registration that supports it (Server #1017). + let (status, registered) = request(&app, "POST", "/api/worker/register", json!({"worker_id":"continuation-worker", + "task_queue":"test","runtime":"php","build_id":"build-1","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":["sticky_execution"]})).await; + assert_eq!(status, StatusCode::CREATED, "{registered}"); let initial=request(&app,"POST","/api/workflows",json!({"workflow_id":"continuation-original","workflow_type":"echo", "task_queue":"test","input":envelope(Payload::Array(vec![Payload::Long(9007199254740993)])), "execution_timeout_seconds":1200,"run_timeout_seconds":600,"memo":{"keep":{"integer":9007199254740993_i64,"optional":null,"unicode":"continue λ"},"empty":[]}})).await; diff --git a/rust/tests/execution/sticky_execution.rs b/rust/tests/execution/sticky_execution.rs new file mode 100644 index 000000000..358182666 --- /dev/null +++ b/rust/tests/execution/sticky_execution.rs @@ -0,0 +1,100 @@ +//! Server #1017: sticky execution as published PHP Server 2.11.8 runs it. +//! A completion's `sticky_cache` claims the run for its worker; each later +//! workflow claim reports the replay that worker should expect. History is +//! still delivered in full. +use super::*; + +async fn sticky_worker(app: &Router, worker: &str) { + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":worker,"task_queue":"test", + "runtime":"php","build_id":"build-1","supported_workflow_types":["echo"],"supported_activity_types":[], + "capabilities":["sticky_execution"]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +fn claim(task: &Value, worker: &str, ttl: i64) -> Value { + json!({"worker_id":worker,"workflow_id":task["workflow_id"],"run_id":task["run_id"],"build_id":"build-1", + "ttl_seconds":ttl,"metrics":{"hit":0,"miss":1,"eviction":0,"forced_cold_replay":0}}) +} + +async fn complete_sticky(app: &Router, task: &Value, sticky: Value) -> (StatusCode, Value) { + let mut body = completion(task, json!([{"type":"open_signal_wait","signal_name":"finish"}])); + body["sticky_cache"] = sticky; + request(app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await +} + +async fn signal(app: &Router, workflow: &str) { + let (status, body) = request(app, "POST", &format!("/api/workflows/{workflow}/signal/finish"), + json!({"input":envelope(Payload::Array(vec![Payload::Long(1)]))})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); +} + +#[tokio::test] +async fn a_sticky_claim_tells_each_later_claim_which_replay_to_expect() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let (_, cluster) = request(&app, "GET", "/api/cluster/info", Value::Null).await; + assert_eq!(cluster["worker_protocol"]["server_capabilities"]["sticky_execution"]["supported"], true, "{cluster}"); + sticky_worker(&app, "sticky-a").await; + sticky_worker(&app, "sticky-b").await; + start(&app, "sticky").await; + + let first = poll(&app, "sticky-a", "workflow").await; + assert_eq!(first["sticky_replay_mode"], "cold_replay", "{first}"); + assert_eq!(first["sticky_worker_id"], Value::Null, "{first}"); + // A claim naming another build is refused whole. + let mut wrong = claim(&first, "sticky-a", 300); + wrong["build_id"] = json!("build-2"); + let (status, body) = complete_sticky(&app, &first, wrong).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "sticky_cache_identity_mismatch", "{body}"); + assert_eq!(body["capability"], "sticky_execution", "{body}"); + let (status, body) = complete_sticky(&app, &first, claim(&first, "sticky-a", 300)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + + // The sticky worker expects a hit; the full history still arrives. + signal(&app, "sticky").await; + let second = poll(&app, "sticky-a", "workflow").await; + assert_eq!(second["sticky_replay_mode"], "sticky_hit_expected", "{second}"); + assert_eq!(second["sticky_worker_id"], "sticky-a", "{second}"); + assert!(second["sticky_until"].is_string(), "{second}"); + let events = second["history_events"].as_array().unwrap(); + assert!(matches!(events[0]["event_type"].as_str(), Some("StartAccepted" | "WorkflowStarted")), "{second}"); + let (status, body) = complete_sticky(&app, &second, claim(&second, "sticky-a", 300)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + + // Another worker may claim, with a forced cold replay. + signal(&app, "sticky").await; + let third = poll(&app, "sticky-b", "workflow").await; + assert_eq!(third["sticky_replay_mode"], "forced_cold_replay", "{third}"); + let (status, body) = complete_sticky(&app, &third, claim(&third, "sticky-b", 300)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + + let (_, metrics) = request(&app, "GET", "/api/system/operator-metrics", Value::Null).await; + let sticky = &metrics["operator_metrics"]["sticky_execution"]; + assert_eq!(sticky["enabled"], true, "{sticky}"); + assert_eq!(sticky["active_sticky_runs"], 1, "{sticky}"); + assert_eq!(sticky["hit_expected_last_minute"], 1, "{sticky}"); + assert_eq!(sticky["forced_cold_replay_last_minute"], 1, "{sticky}"); + assert_eq!(sticky["cold_replay_last_minute"], 1, "{sticky}"); + + // Re-registering the sticky worker's process drops its affinity. + sticky_worker(&app, "sticky-b").await; + signal(&app, "sticky").await; + let fourth = poll(&app, "sticky-b", "workflow").await; + assert_eq!(fourth["sticky_replay_mode"], "cold_replay", "{fourth}"); +} + +#[tokio::test] +async fn an_expired_claim_forces_a_cold_replay() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + sticky_worker(&app, "sticky-a").await; + start(&app, "expired").await; + let first = poll(&app, "sticky-a", "workflow").await; + let (status, body) = complete_sticky(&app, &first, claim(&first, "sticky-a", 1)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + tokio::time::sleep(std::time::Duration::from_millis(1100)).await; + signal(&app, "expired").await; + let second = poll(&app, "sticky-a", "workflow").await; + assert_eq!(second["sticky_replay_mode"], "forced_cold_replay", "{second}"); +}