From bd97c3133e9167a7777e06a898a501072ff1c44f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 15:39:01 +0000 Subject: [PATCH] Admit worker registrations like Server 2.10.0 Registration now runs PHP's Laravel request rules, with the capability manifest required from protocol 1.18, and then PHP's capability guards in order: prepared local activities, Nexus and cooperative cancellation, the manifest's flag and floor agreement, workflow metadata capabilities and message streams. Refusals carry PHP's fields plus protocol_version and server_capabilities. An unnamed worker gets a ULID, missing type lists are stored empty, and the manifest is stored normalized. The rule engine gains numeric and required_array_keys, and a nested wildcard rule covers only array items, as Laravel's expansion does. The execution tests' router adds the manifest an SDK sends; the new admission tests use the server's own router. --- .../2026-10-22-worker-admission.md | 60 +++++ docs/rust-server-port.md | 3 - rust/src/runtime/http.rs | 59 +++-- rust/src/runtime/laravel_rules.rs | 33 ++- rust/src/runtime/mod.rs | 1 + rust/src/runtime/worker_admission.rs | 228 ++++++++++++++++++ rust/tests/execution.rs | 49 +++- rust/tests/execution/worker_admission.rs | 161 +++++++++++++ 8 files changed, 564 insertions(+), 30 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-22-worker-admission.md create mode 100644 rust/src/runtime/worker_admission.rs create mode 100644 rust/tests/execution/worker_admission.rs diff --git a/docs/rust-server-port-checkpoints/2026-10-22-worker-admission.md b/docs/rust-server-port-checkpoints/2026-10-22-worker-admission.md new file mode 100644 index 000000000..c30d1653a --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-22-worker-admission.md @@ -0,0 +1,60 @@ +# October 22, worker registration admission + +- **Gap.** Native accepted a registration without the `capability_manifest` + PHP requires from protocol 1.18. It also skipped most of PHP's request + rules and capability guards, refused a missing `worker_id` or type list, + and answered metadata-capability mismatches in its own shape. +- **Source.** Server 2.10.0 `WorkerController::register`: + - its Laravel rules; + - the prepared-local-activity, Nexus and cooperative-cancellation guards; + - `guardPortableWorkerCapabilityManifest`; + - `WorkflowMetadataCapabilityPolicy::firstProtocolMismatch`; + - the message-streams floor; + - `portableWorkerCapabilityManifest`. +- **Request rules.** Laravel trims strings and turns empty ones into null + first; native now does the same for the whole body. + - Every field follows PHP's rules: `runtime` from PHP's list, sizes, + capacities, task slots, process metrics, heartbeat interval. + - `capability_manifest` is required from protocol 1.18 and must hold the + `local_activities`, `worker_sessions` and `sticky_execution` entries. + Each entry needs `supported` (boolean) and `minimum_protocol_version`. + - Failures answer Laravel's 422 `validation_failed` with its messages. The + rule engine gained `numeric` and `required_array_keys`. A `*.field` rule + now covers only array items, as Laravel's expansion does. +- **Guards.** Each answers 409 with `registered: false` and PHP's fields, in + PHP's order: + - `prepared_local_activity_cancellation_policy_capability_mismatch`, + `prepared_local_activity_group_capability_mismatch` and + `prepared_local_activity_capability_mismatch`; + - `nexus_cooperative_cancellation_capability_mismatch` and + `cooperative_cancellation_protocol_mismatch`, below protocol 1.20; + - `worker_capability_manifest_mismatch`, when a flat affinity capability + and its `supported` flag disagree; + - `worker_capability_protocol_floor_mismatch`, when an entry's floor is + below 1.18 or above the request's protocol; + - `workflow_metadata_capability_protocol_mismatch`, for `memo_upserts` + below 1.14 and `typed_search_attributes` below 1.16. This now uses PHP's + `capability`, `requested_version` and `remediation` fields, not native's + former `worker_capability` and `recorded`; + - `message_streams_unavailable`, below 1.15. +- **Accepted registrations.** + - A missing `worker_id` gets a ULID, as PHP does. + - Missing type lists are stored empty. + - Capabilities keep their non-empty strings. + - The manifest is stored normalized: the three entries, a trimmed floor, + and non-empty `implementation` and `reason`. +- **Worker protocol bodies.** Registration refusals now carry + `protocol_version` and `server_capabilities`, as `WorkerProtocol::json` + does. Other worker-plane refusals still lack them (#46's shape sweep). +- **Tests.** + - `worker_admission.rs` uses the server's own router. It covers the + manifest requirement and its errors, field rules with Laravel's + messages, the generated id, the stored manifest, and every guard. + - The other tests' router adds the manifest an SDK sends from 1.18, + matching each registration's flat capabilities. + - Two cooperative tests registered `cooperative_cancellation` at protocol + 1.19, which PHP refuses. They now register at 1.20 and still poll at the + protocol each case tests. + - Full SQLite run (all targets): 230 unit, 401 execution and 2 upgrade + tests passed. Worker, registration, admission, cooperative and + namespace tests: PostgreSQL 124, MySQL 124. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index ab3ab1fbd..ad722d11a 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -171,9 +171,6 @@ contract fixture before its correction. deep. Native checks every body as PHP does before any route, but its handlers parse with serde_json's 128-level limit. A body nested 129 to 512 levels deep passes the check and is then refused with a plain-text 400. -- **Registration without a capability manifest.** At protocol 1.19, - published PHP 2.5.14 refuses a registration that has no - `capability_manifest` (422). Native accepts it. - **Python worker shutdown after a late synchronous query handler (not yet diagnosed).** In the SQLite query-authority cohort on #515's run 39431491496, the published Python SDK 2.5.15 client in `late-handler-sync` diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 268db087f..95051680c 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -1319,29 +1319,46 @@ async fn register( headers: HeaderMap, Json(body): Json, ) -> Result<(StatusCode, Json)> { - if body["capabilities"] - .as_array() - .is_some_and(|flags| flags.iter().any(|flag| flag == "memo_upserts")) - { - memo_protocol(&headers)?; + let protocol = request_protocol(&headers)?; + // Laravel trims every string and turns empty ones into null first. + let mut body = super::laravel_rules::normalize(&body); + super::worker_admission::admit(&body, Some(protocol.trim())).map_err(worker_refusal)?; + // PHP names an unnamed worker with a fresh ULID and stores the + // normalized capability manifest. + if body["worker_id"].is_null() { + body["worker_id"] = json!(ulid::Ulid::new().to_string()); } - if body["capabilities"].as_array().is_some_and(|flags| { - flags - .iter() - .any(|flag| flag == super::search_attribute_upserts::CAPABILITY) - }) && !protocol_at_least(&headers, 16)? - { - return Err(super::RuntimeError::Protocol { - status: StatusCode::CONFLICT, - response: json!({"reason":"workflow_metadata_capability_protocol_mismatch", - "worker_capability":super::search_attribute_upserts::CAPABILITY, - "minimum_protocol_version":super::search_attribute_upserts::MINIMUM_PROTOCOL_VERSION,"recorded":false}), - }); + let manifest = super::worker_admission::manifest(&body); + body["capability_manifest"] = manifest; + // Missing type lists are empty; capabilities keep their non-empty + // strings (`nonEmptyStringArray`). + for field in ["supported_workflow_types", "supported_activity_types"] { + if body[field].is_null() { + body[field] = json!([]); + } + } + let flags: Vec = body["capabilities"] + .as_array() + .into_iter() + .flatten() + .filter(|flag| flag.as_str().is_some_and(|flag| !flag.is_empty())) + .cloned() + .collect(); + body["capabilities"] = json!(flags); + let registered = runtime.register(body, Some(protocol)).await.map_err(worker_refusal)?; + Ok((StatusCode::CREATED, Json(worker_response(registered)))) +} + +/// `WorkerProtocol::json` names the protocol and server capabilities on +/// every worker response, refusals included. +fn worker_refusal(error: super::RuntimeError) -> super::RuntimeError { + match error { + super::RuntimeError::Protocol { status, response } => super::RuntimeError::Protocol { + status, + response: worker_response(response), + }, + other => other, } - Ok(( - StatusCode::CREATED, - Json(worker_response(runtime.register(body, Some(request_protocol(&headers)?)).await?)), - )) } fn worker_response(mut body: Value) -> Value { diff --git a/rust/src/runtime/laravel_rules.rs b/rust/src/runtime/laravel_rules.rs index 66cc2a192..7c32062cb 100644 --- a/rust/src/runtime/laravel_rules.rs +++ b/rust/src/runtime/laravel_rules.rs @@ -25,7 +25,11 @@ pub(crate) enum Rule { String, Array, Integer, + /// `numeric`: PHP's `is_numeric`. + Numeric, Boolean, + /// `required_array_keys`: an array holding each named key. + RequiredArrayKeys(&'static [&'static str]), /// A size floor: a number, string length or item count, by type rule. Min(i64), Max(i64), @@ -97,7 +101,12 @@ fn expand(data: &Value, pattern: &str) -> Vec { Some(Value::Object(fields)) => next.extend(fields.keys().map(|key| joined(key))), _ => {} } - } else { + } else if prefix.is_empty() + || !pattern_has_wildcard_before(pattern, &prefix) + || lookup(data, &prefix).is_some_and(|item| item.is_array() || item.is_object()) + { + // Laravel initializes `item.field` only under array items + // (`data_set` with `*`), so a scalar item has no such field. next.push(joined(segment)); } } @@ -106,6 +115,12 @@ fn expand(data: &Value, pattern: &str) -> Vec { paths } +/// Whether `prefix`, a path built from `pattern`, already passed a `*`. +fn pattern_has_wildcard_before(pattern: &str, prefix: &str) -> bool { + let depth = prefix.split('.').count(); + pattern.split('.').take(depth).any(|segment| segment == "*") +} + /// Laravel's displayable attribute: expanded `*` attributes keep their raw /// path; others read with spaces for underscores. fn display(attribute: &str, expanded: bool) -> String { @@ -163,7 +178,7 @@ fn php_string(value: &Value) -> Option { } fn message(rule: Rule, attribute: &str, rules: &[Rule], value: Option<&Value>) -> String { - let kind = if rules.iter().any(|rule| matches!(rule, Rule::Integer)) { + let kind = if rules.iter().any(|rule| matches!(rule, Rule::Integer | Rule::Numeric)) { "numeric" } else if rules.iter().any(|rule| matches!(rule, Rule::Array)) || value.is_some_and(Value::is_array) { "array" @@ -178,6 +193,10 @@ fn message(rule: Rule, attribute: &str, rules: &[Rule], value: Option<&Value>) - Rule::String => format!("The {attribute} field must be a string."), Rule::Array => format!("The {attribute} field must be an array."), Rule::Integer => format!("The {attribute} field must be an integer."), + Rule::Numeric => format!("The {attribute} field must be a number."), + Rule::RequiredArrayKeys(keys) => { + format!("The {attribute} field must contain entries for: {}.", keys.join(", ")) + } Rule::Boolean => format!("The {attribute} field must be true or false."), Rule::In(_) => format!("The selected {attribute} is invalid."), Rule::Min(limit) => match kind { @@ -199,7 +218,7 @@ fn message(rule: Rule, attribute: &str, rules: &[Rule], value: Option<&Value>) - /// Laravel's `getSize`: a numeric value under a numeric rule, otherwise /// an item count or the string length. fn size(value: &Value, rules: &[Rule]) -> Option { - if rules.iter().any(|rule| matches!(rule, Rule::Integer)) { + if rules.iter().any(|rule| matches!(rule, Rule::Integer | Rule::Numeric)) { match value { Value::Number(number) => return number.as_f64(), Value::String(text) if super::schedule_cron::php_is_numeric(text) => { @@ -231,6 +250,14 @@ fn passes(rule: Rule, data: &Value, value: Option<&Value>, rules: &[Rule]) -> bo .is_some_and(|text| super::schedule_backfill::php_date(text).is_some()), Rule::Array => value.is_some_and(|value| value.is_array() || value.is_object()), Rule::Integer => value.and_then(integer).is_some(), + Rule::Numeric => value.is_some_and(|value| match value { + Value::Number(_) => true, + Value::String(text) => super::schedule_cron::php_is_numeric(text), + _ => false, + }), + Rule::RequiredArrayKeys(keys) => value + .and_then(Value::as_object) + .is_some_and(|fields| keys.iter().all(|key| fields.contains_key(*key))), Rule::Boolean => matches!( value, Some(Value::Bool(_)) diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index d6dd33a1e..a4dc48bf7 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -102,6 +102,7 @@ mod wait_timeouts; mod worker_contracts; mod worker_incarnations; mod worker_management; +mod worker_admission; mod workflow_failures; mod workflow_task_guard; diff --git a/rust/src/runtime/worker_admission.rs b/rust/src/runtime/worker_admission.rs new file mode 100644 index 000000000..e8ce3f6ae --- /dev/null +++ b/rust/src/runtime/worker_admission.rs @@ -0,0 +1,228 @@ +//! Server 2.10.0 `WorkerController::register` admission: Laravel's request +//! rules, then the capability guards, in PHP's order. A passing request +//! continues to the store with the normalized capability manifest. +use super::laravel_rules::{self, Rule, Rule::*}; +use super::{Result, RuntimeError}; +use axum::http::StatusCode; +use serde_json::{Map, Value, json}; + +/// `WorkerProtocol::PORTABLE_WORKER_AFFINITY_CAPABILITIES`. +const AFFINITY: &[&str] = &["local_activities", "worker_sessions", "sticky_execution"]; +/// `WorkerProtocol::PORTABLE_WORKER_AFFINITY_MINIMUM_PROTOCOL_VERSION`. +const AFFINITY_FLOOR: &str = "1.18"; +const COOPERATIVE: &str = "cooperative_cancellation"; +/// `CooperativeCancellationPolicy` and `PreparedLocalActivityPolicy`. +const COOPERATIVE_FLOOR: &str = "1.20"; +const PREPARED: &str = "prepared_local_activities"; +const PREPARED_GROUPS: &str = "prepared_local_activity_groups"; +const PREPARED_POLICIES: &str = "prepared_local_activity_cancellation_policies"; +const NEXUS_COOPERATIVE: &str = "nexus_cooperative_cancellation"; +/// `WorkflowMetadataCapabilityPolicy::definitions`, in order. +const METADATA: &[(&str, &str)] = &[("memo_upserts", "1.14"), ("typed_search_attributes", "1.16")]; +/// `MessageStreamsContract`. +const MESSAGE_STREAMS: (&str, &str) = ("message_streams", "1.15"); + +const RUNTIMES: &[&str] = &["php", "python", "rust", "typescript", "go", "java", "external"]; +const MANIFEST_REQUIRED: &[Rule] = &[Required, Array, RequiredArrayKeys(AFFINITY)]; +const MANIFEST_NULLABLE: &[Rule] = &[Nullable, Array, RequiredArrayKeys(AFFINITY)]; + +/// The request rules, with the manifest required from protocol 1.18 +/// (`WorkerProtocol::portableWorkerAffinitySupported`). +fn rules(manifest: &'static [Rule]) -> Vec<(&'static str, &'static [Rule])> { + vec![ + ("worker_id", &[Nullable, String, Max(255)]), + ("task_queue", &[Required, String, Max(255)]), + ("runtime", &[Required, String, In(RUNTIMES)]), + ("sdk_version", &[Nullable, String, Max(64)]), + ("build_id", &[Nullable, String, Max(255)]), + ("supported_workflow_types", &[Nullable, Array]), + ("supported_workflow_types.*", &[String]), + ("workflow_definition_fingerprints", &[Nullable, Array]), + ("workflow_definition_fingerprints.*", &[String, Max(255)]), + ("workflow_command_contracts", &[Nullable, Array]), + ("supported_activity_types", &[Nullable, Array]), + ("supported_activity_types.*", &[String]), + ("capabilities", &[Nullable, Array]), + ("capabilities.*", &[String, Max(255)]), + ("capability_manifest", manifest), + ("capability_manifest.*", &[Array]), + ("capability_manifest.*.supported", &[Required, Boolean]), + ("capability_manifest.*.minimum_protocol_version", &[Required, String, Max(32)]), + ("capability_manifest.*.implementation", &[Nullable, String, Max(255)]), + ("capability_manifest.*.reason", &[Nullable, String, Max(255)]), + ("max_concurrent_workflow_tasks", &[Nullable, Integer, Min(0)]), + ("max_concurrent_activity_tasks", &[Nullable, Integer, Min(0)]), + ("max_concurrent_worker_sessions", &[Nullable, Integer, Min(1)]), + ("task_slots", &[Nullable, Array]), + ("task_slots.workflow_available", &[Nullable, Integer, Min(0)]), + ("task_slots.activity_available", &[Nullable, Integer, Min(0)]), + ("task_slots.session_available", &[Nullable, Integer, Min(0)]), + ("process_metrics", &[Nullable, Array]), + ("process_metrics.cpu_percent", &[Nullable, Numeric, Min(0)]), + ("process_metrics.memory_bytes", &[Nullable, Integer, Min(0)]), + ("process_metrics.process_uptime_seconds", &[Nullable, Integer, Min(0)]), + ("process_metrics.process_id", &[Nullable, Integer, Min(0)]), + ("process_metrics.host", &[Nullable, String, Max(255)]), + ("process_metrics.process_started_at", &[Nullable, String, Max(64)]), + ("heartbeat_interval_seconds", &[Nullable, Integer, Min(1), Max(3600)]), + ] +} + +/// PHP's `version_compare($version, $minimum, '>=')` on dotted numbers; +/// a missing version meets nothing (`WorkerProtocol::versionMeetsMinimum`). +pub(super) fn at_least(version: Option<&str>, minimum: &str) -> bool { + let parts = |text: &str| -> Vec { text.trim().split('.').map(|part| part.parse().unwrap_or(0)).collect() }; + let Some(version) = version.filter(|version| !version.trim().is_empty()) else { + return false; + }; + let (version, minimum) = (parts(version), parts(minimum)); + let length = version.len().max(minimum.len()); + let padded = |list: &[u64]| (0..length).map(|index| list.get(index).copied().unwrap_or(0)).collect::>(); + padded(&version) >= padded(&minimum) +} + +fn refused(mut response: Value) -> RuntimeError { + response["registered"] = json!(false); + RuntimeError::Protocol { status: StatusCode::CONFLICT, response } +} + +/// The registration's flat capabilities, as PHP's `nonEmptyStringArray`. +fn capabilities(body: &Value) -> Vec<&str> { + body["capabilities"] + .as_array() + .into_iter() + .flatten() + .filter_map(Value::as_str) + .filter(|capability| !capability.is_empty()) + .collect() +} + +/// `portableWorkerCapabilityManifest`: the three affinity entries that are +/// arrays, with `supported` and a trimmed floor, plus non-empty +/// `implementation` and `reason`. +pub(super) fn manifest(body: &Value) -> Value { + let mut normalized = Map::new(); + for capability in AFFINITY { + let Some(entry) = body["capability_manifest"].get(*capability).filter(|entry| entry.is_object()) else { + continue; + }; + let supported = match &entry["supported"] { + Value::Bool(flag) => *flag, + Value::Number(number) => number.as_f64().is_some_and(|number| number != 0.0), + Value::String(text) => !text.is_empty() && text != "0", + _ => false, + }; + let floor = match &entry["minimum_protocol_version"] { + Value::String(text) => text.trim().to_owned(), + Value::Number(number) => number.to_string(), + _ => std::string::String::new(), + }; + let mut item = json!({"supported": supported, "minimum_protocol_version": floor}); + for field in ["implementation", "reason"] { + if let Some(text) = entry[field].as_str().map(str::trim).filter(|text| !text.is_empty()) { + item[field] = json!(text); + } + } + normalized.insert((*capability).to_owned(), item); + } + Value::Object(normalized) +} + +/// The admission checks before any registration state is read. `body` is +/// the request after Laravel's trimming; `version` is the request's +/// worker protocol header. +pub(super) fn admit(body: &Value, version: Option<&str>) -> Result<()> { + let manifest_rule = if at_least(Some(version.unwrap_or(super::WORKER_PROTOCOL_VERSION)), AFFINITY_FLOOR) { + MANIFEST_REQUIRED + } else { + MANIFEST_NULLABLE + }; + laravel_rules::validate(body, &rules(manifest_rule))?; + let flags = capabilities(body); + let has = |capability: &str| flags.contains(&capability); + let cooperative = has(COOPERATIVE) && at_least(version, COOPERATIVE_FLOOR); + let requested = json!(version); + let pair = json!([COOPERATIVE, PREPARED]); + if has(PREPARED_POLICIES) && !(cooperative && has(PREPARED) && has(PREPARED_POLICIES)) { + return Err(refused(json!({"reason": "prepared_local_activity_cancellation_policy_capability_mismatch", + "required_capabilities": pair, "minimum_protocol_version": COOPERATIVE_FLOOR, "requested_version": requested}))); + } + if has(PREPARED_GROUPS) && (!has(PREPARED) || !cooperative) { + return Err(refused(json!({"reason": "prepared_local_activity_group_capability_mismatch", + "required_capabilities": pair, "minimum_protocol_version": COOPERATIVE_FLOOR, "requested_version": requested}))); + } + if has(PREPARED) && !cooperative { + return Err(refused(json!({"reason": "prepared_local_activity_capability_mismatch", + "required_capability": COOPERATIVE, "minimum_protocol_version": COOPERATIVE_FLOOR, "requested_version": requested}))); + } + if has(NEXUS_COOPERATIVE) && !cooperative { + return Err(refused(json!({"reason": "nexus_cooperative_cancellation_capability_mismatch", + "required_capabilities": [COOPERATIVE, NEXUS_COOPERATIVE], "minimum_protocol_version": COOPERATIVE_FLOOR, + "requested_version": requested}))); + } + if has(COOPERATIVE) && !cooperative { + return Err(refused(json!({"reason": "cooperative_cancellation_protocol_mismatch", + "minimum_protocol_version": COOPERATIVE_FLOOR, "requested_version": requested}))); + } + // `guardPortableWorkerCapabilityManifest`: the structured flag and the + // flat capability agree, and an advertised capability meets its floor. + let normalized = manifest(body); + for capability in AFFINITY { + let advertised = has(capability); + let entry = normalized.get(*capability); + if entry.is_none() && !advertised { + continue; + } + if entry.is_none_or(|entry| entry["supported"].as_bool() != Some(advertised)) { + return Err(refused(json!({"reason": "worker_capability_manifest_mismatch", "capability": capability, + "requested_version": requested, "minimum_protocol_version": AFFINITY_FLOOR, + "remediation": "Keep the structured supported flag and the flat routing capability in exact agreement."}))); + } + let floor = entry.and_then(|entry| entry["minimum_protocol_version"].as_str()).unwrap_or_default(); + if floor.is_empty() || !at_least(Some(floor), AFFINITY_FLOOR) || (advertised && !at_least(version, floor)) { + return Err(refused(json!({"reason": "worker_capability_protocol_floor_mismatch", "capability": capability, + "requested_version": requested, "minimum_protocol_version": AFFINITY_FLOOR, + "remediation": format!("Advertise {capability} only from a worker targeting protocol {AFFINITY_FLOOR} or newer.")}))); + } + } + if let Some((capability, minimum)) = + METADATA.iter().find(|(capability, minimum)| has(capability) && !at_least(version, minimum)) + { + return Err(refused(json!({"reason": "workflow_metadata_capability_protocol_mismatch", + "capability": capability, "requested_version": requested, "minimum_protocol_version": minimum, + "remediation": format!("Advertise {capability} only while sending worker protocol {minimum} or newer.")}))); + } + let (streams, minimum) = MESSAGE_STREAMS; + if has(streams) && !at_least(version, minimum) { + return Err(refused(json!({"reason": "message_streams_unavailable", "requested_version": requested, + "minimum_protocol_version": minimum, + "remediation": format!("Advertise message_streams only while sending worker protocol {minimum} or newer.")}))); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn protocol_versions_compare_like_php() { + assert!(at_least(Some("1.20"), "1.18")); + assert!(at_least(Some("1.18"), "1.18")); + assert!(!at_least(Some("1.9"), "1.18")); + assert!(at_least(Some("2.0"), "1.20")); + assert!(!at_least(None, "1.0")); + } + + #[test] + fn the_manifest_keeps_the_affinity_entries_that_are_arrays() { + let body = json!({"capability_manifest": { + "local_activities": {"supported": 1, "minimum_protocol_version": " 1.18 ", "implementation": " sdk ", "reason": ""}, + "worker_sessions": "unsupported", + "sticky_execution": {"supported": false, "minimum_protocol_version": "1.18"}, + "other": {"supported": true, "minimum_protocol_version": "1.18"}}}); + assert_eq!(manifest(&body), json!({ + "local_activities": {"supported": true, "minimum_protocol_version": "1.18", "implementation": "sdk"}, + "sticky_execution": {"supported": false, "minimum_protocol_version": "1.18"}})); + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 6ac4387dc..b45ee5b68 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -5,7 +5,7 @@ use axum::{ }; use durable_workflow_server::{ codec::{Value as Payload, ValueCodec}, - runtime::{Runtime, router}, + runtime::Runtime, }; use serde_json::{Value, json}; use tower::ServiceExt; @@ -99,6 +99,8 @@ mod cooperative_waits; mod child_policies; #[path = "execution/definition_routing.rs"] mod definition_routing; +#[path = "execution/worker_admission.rs"] +mod worker_admission; /// `WorkflowDefinitionRouting::ENTRY_VALIDATION_CAPABILITY`. const ENTRY_VALIDATION: &str = "workflow_entry_arguments_validation"; #[path = "execution/parent_close_policies.rs"] @@ -3192,8 +3194,10 @@ async fn cooperative_root_request_requires_original_capable_claim_not_later_flee let registration = json!({"worker_id":"cooperative-owner","task_queue":"test","runtime":"php", "supported_workflow_types":["echo"],"supported_activity_types":[], "capabilities":if original_capable {json!(["cooperative_cancellation"])} else {json!([])}}); + // Registration needs protocol 1.20 to advertise cooperative + // cancellation; the claim records the protocol of the poll. assert_eq!( - request(&app, "POST", "/api/worker/register", registration.clone()) + request_protocol(&app, "POST", "/api/worker/register", registration.clone(), "1.20") .await .0, StatusCode::CREATED @@ -3303,7 +3307,7 @@ async fn cooperative_requested_root_ignores_protocol_spoofing_and_requires_capab let mut capable = registration; capable["capabilities"] = json!(["cooperative_cancellation"]); assert_eq!( - request(&app, "POST", "/api/worker/register", capable) + request_protocol(&app, "POST", "/api/worker/register", capable, "1.20") .await .0, StatusCode::CREATED @@ -5091,3 +5095,42 @@ async fn authentication_precedes_protocol_and_namespace_checks() { runtime.close().await; database.remove().await; } + +/// The server router, behind an SDK's registration habit: from protocol +/// 1.18 an SDK sends a `capability_manifest` that agrees with its flat +/// capabilities, so one is added where a test registration omits it. +/// Tests of the manifest rules use `durable_workflow_server::runtime::router`. +fn router(runtime: Runtime) -> Router { + durable_workflow_server::runtime::router(runtime).layer(axum::middleware::from_fn(sdk_capability_manifest)) +} + +async fn sdk_capability_manifest(request: Request, next: axum::middleware::Next) -> axum::response::Response { + let floor = request + .headers() + .get("x-durable-workflow-protocol-version") + .and_then(|version| version.to_str().ok()) + .and_then(|version| version.trim().strip_prefix("1.")) + .and_then(|minor| minor.parse::().ok()) + .is_some_and(|minor| minor >= 18); + if request.uri().path() != "/api/worker/register" || !floor { + return next.run(request).await; + } + let (parts, body) = request.into_parts(); + let bytes = to_bytes(body, usize::MAX).await.unwrap(); + let bytes = match serde_json::from_slice::(&bytes) { + Ok(Value::Object(mut fields)) if !fields.contains_key("capability_manifest") => { + let flags = fields.get("capabilities").cloned().unwrap_or(Value::Null); + let advertised = |capability: &str| flags.as_array().is_some_and(|flags| flags.iter().any(|flag| flag == capability)); + let manifest: serde_json::Map = ["local_activities", "worker_sessions", "sticky_execution"] + .into_iter() + .map(|capability| { + (capability.to_owned(), json!({"supported": advertised(capability), "minimum_protocol_version": "1.18"})) + }) + .collect(); + fields.insert("capability_manifest".into(), Value::Object(manifest)); + Value::Object(fields).to_string().into_bytes() + } + _ => bytes.to_vec(), + }; + next.run(Request::from_parts(parts, Body::from(bytes))).await +} diff --git a/rust/tests/execution/worker_admission.rs b/rust/tests/execution/worker_admission.rs new file mode 100644 index 000000000..8b7ffda77 --- /dev/null +++ b/rust/tests/execution/worker_admission.rs @@ -0,0 +1,161 @@ +use super::*; + +// Server 2.10.0 `WorkerController::register`: Laravel's request rules, with +// `capability_manifest` required from protocol 1.18, then the capability +// guards in PHP's order. These tests use the server's own router, without +// the SDK-style manifest the other tests' router adds. + +async fn app() -> (TestDatabase, Router) { + let database = TestDatabase::new().await; + let app = durable_workflow_server::runtime::router(database.open().await.unwrap()); + (database, app) +} + +fn affinity(local: bool, sessions: bool, sticky: bool) -> Value { + let entry = |supported: bool| json!({"supported": supported, "minimum_protocol_version": "1.18"}); + json!({"local_activities": entry(local), "worker_sessions": entry(sessions), "sticky_execution": entry(sticky)}) +} + +async fn register_at(app: &Router, version: &str, fields: Value) -> (StatusCode, Value) { + let mut body = json!({"worker_id": "admitted", "task_queue": "test", "runtime": "php"}); + for (field, value) in fields.as_object().unwrap() { + body[field] = value.clone(); + } + request_protocol(app, "POST", "/api/worker/register", body, version).await +} + +#[tokio::test] +async fn the_manifest_is_required_from_protocol_1_18() { + let (_database, app) = app().await; + let (status, body) = register_at(&app, "1.19", json!({})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["reason"], "validation_failed"); + assert_eq!(body["errors"]["capability_manifest"], json!(["The capability manifest field is required."]), "{body}"); + // `WorkerProtocol::json` names the protocol on refusals too. + assert_eq!(body["protocol_version"], "1.20"); + assert!(body["server_capabilities"].is_object(), "{body}"); + + let (status, body) = register_at(&app, "1.17", json!({})).await; + assert_eq!(status, StatusCode::CREATED, "an older worker needs no manifest: {body}"); + + let (status, body) = register_at(&app, "1.19", json!({"capability_manifest": {"local_activities": {}}})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!( + body["errors"], + json!({ + "capability_manifest": ["The capability manifest field must contain entries for: local_activities, worker_sessions, sticky_execution."], + "capability_manifest.local_activities.supported": ["The capability_manifest.local_activities.supported field is required."], + "capability_manifest.local_activities.minimum_protocol_version": ["The capability_manifest.local_activities.minimum_protocol_version field is required."], + }), + "{body}" + ); + + // A scalar entry is not an array and has no fields to require. + let mut manifest = affinity(false, false, false); + manifest["worker_sessions"] = json!("no"); + let (status, body) = register_at(&app, "1.19", json!({"capability_manifest": manifest})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["errors"], json!({"capability_manifest.worker_sessions": ["The capability_manifest.worker_sessions field must be an array."]}), "{body}"); +} + +#[tokio::test] +async fn registration_fields_follow_php_rules() { + let (_database, app) = app().await; + let manifest = affinity(false, false, false); + let (status, body) = register_at(&app, "1.19", json!({"capability_manifest": manifest, "runtime": "cobol", + "process_metrics": {"cpu_percent": "high"}, "heartbeat_interval_seconds": 0})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!( + body["errors"], + json!({ + "runtime": ["The selected runtime is invalid."], + "process_metrics.cpu_percent": ["The process metrics.cpu percent field must be a number."], + "heartbeat_interval_seconds": ["The heartbeat interval seconds field must be at least 1."], + }), + "{body}" + ); + + // An unnamed worker gets a ULID; strings are trimmed and the manifest is + // stored normalized. + let mut manifest = affinity(false, false, false); + manifest["local_activities"]["implementation"] = json!(" sdk "); + manifest["local_activities"]["extra"] = json!("dropped"); + let (status, body) = register_at(&app, "1.19", json!({"worker_id": "", "task_queue": " test ", "capability_manifest": manifest})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + let worker_id = body["worker_id"].as_str().unwrap(); + assert_eq!(worker_id.len(), 26, "{body}"); + let (status, described) = request(&app, "GET", &format!("/api/workers/{worker_id}"), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{described}"); + assert_eq!(described["task_queue"], "test", "{described}"); + assert_eq!( + described["capability_manifest"]["local_activities"], + json!({"supported": false, "minimum_protocol_version": "1.18", "implementation": "sdk"}), + "{described}" + ); +} + +#[tokio::test] +async fn capability_guards_refuse_like_php() { + let (_database, app) = app().await; + let manifest = affinity(false, false, false); + let refused = |capabilities: Value, version: &'static str, manifest: Value| { + let app = app.clone(); + async move { + let (status, body) = + register_at(&app, version, json!({"capabilities": capabilities, "capability_manifest": manifest})).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["registered"], false, "{body}"); + assert_eq!(body["protocol_version"], "1.20", "{body}"); + body + } + }; + + // The flat capability and the structured flag must agree. + let body = refused(json!(["worker_sessions"]), "1.19", manifest.clone()).await; + assert_eq!((body["reason"].as_str(), body["capability"].as_str()), (Some("worker_capability_manifest_mismatch"), Some("worker_sessions")), "{body}"); + let body = refused(json!([]), "1.19", affinity(false, false, true)).await; + assert_eq!(body["capability"], "sticky_execution", "{body}"); + + // An advertised capability needs its floor at 1.18 or later and met. + let mut low = affinity(true, false, false); + low["local_activities"]["minimum_protocol_version"] = json!("1.17"); + let body = refused(json!(["local_activities"]), "1.19", low).await; + assert_eq!(body["reason"], "worker_capability_protocol_floor_mismatch", "{body}"); + let mut high = affinity(true, false, false); + high["local_activities"]["minimum_protocol_version"] = json!("1.21"); + let body = refused(json!(["local_activities"]), "1.20", high).await; + assert_eq!(body["reason"], "worker_capability_protocol_floor_mismatch", "{body}"); + assert_eq!(body["remediation"], "Advertise local_activities only from a worker targeting protocol 1.18 or newer."); + + // Cooperative cancellation and what builds on it need protocol 1.20. + let body = refused(json!(["cooperative_cancellation"]), "1.19", manifest.clone()).await; + assert_eq!(body["reason"], "cooperative_cancellation_protocol_mismatch", "{body}"); + assert_eq!(body["requested_version"], "1.19"); + let body = refused(json!(["prepared_local_activities"]), "1.20", manifest.clone()).await; + assert_eq!(body["reason"], "prepared_local_activity_capability_mismatch", "{body}"); + assert_eq!(body["required_capability"], "cooperative_cancellation"); + let body = refused(json!(["cooperative_cancellation", "prepared_local_activity_groups"]), "1.20", manifest.clone()).await; + assert_eq!(body["reason"], "prepared_local_activity_group_capability_mismatch", "{body}"); + let body = refused(json!(["prepared_local_activity_cancellation_policies"]), "1.20", manifest.clone()).await; + assert_eq!(body["reason"], "prepared_local_activity_cancellation_policy_capability_mismatch", "{body}"); + let body = refused(json!(["nexus_cooperative_cancellation"]), "1.20", manifest.clone()).await; + assert_eq!(body["reason"], "nexus_cooperative_cancellation_capability_mismatch", "{body}"); + + // Workflow metadata and message streams have their own floors. + let body = refused(json!(["memo_upserts"]), "1.13", Value::Null).await; + assert_eq!( + (body["reason"].as_str(), body["capability"].as_str(), body["minimum_protocol_version"].as_str()), + (Some("workflow_metadata_capability_protocol_mismatch"), Some("memo_upserts"), Some("1.14")), + "{body}" + ); + assert_eq!(body["remediation"], "Advertise memo_upserts only while sending worker protocol 1.14 or newer."); + let body = refused(json!(["typed_search_attributes"]), "1.15", Value::Null).await; + assert_eq!(body["capability"], "typed_search_attributes", "{body}"); + let body = refused(json!(["message_streams"]), "1.14", Value::Null).await; + assert_eq!(body["reason"], "message_streams_unavailable", "{body}"); + + // At 1.20 a cooperative worker with matching flags registers. + let (status, body) = register_at(&app, "1.20", json!({"capabilities": ["cooperative_cancellation", "local_activities"], + "capability_manifest": affinity(true, false, false)})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +}