diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index 73fc654b7..c6d194fe2 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -155,6 +155,11 @@ where { let contracts = DB::document_row(&worker, "workflow_command_contracts")?; if let Some(contract) = contracts.get(workflow_type) { + // Older development registrations may predate validation. + // Never turn a malformed entry into authoritative empties. + let Ok(contract) = super::worker_contracts::declaration(contract) else { + continue; + }; for (source, target) in [ ("signals", "declared_signals"), ("signal_contracts", "declared_signal_contracts"), @@ -164,7 +169,9 @@ where ("update_contracts", "declared_update_contracts"), ("update_validators", "declared_update_validators"), ] { - started[target] = contract[source].clone(); + if let Some(value) = contract.get(source) { + started[target] = value.clone(); + } } break; } diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 0d051c048..2fa6aa8fe 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -63,6 +63,7 @@ pub mod upgrade; mod visibility; mod wait_projection; mod wait_timeouts; +mod worker_contracts; mod worker_incarnations; mod workflow_failures; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 2f25c77c3..88da7f9e7 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -813,27 +813,29 @@ where } } let runtime = text(&body, "runtime")?; - let contracts = body + let supplied_contracts = body .get("workflow_command_contracts") - .cloned() - .unwrap_or_else(|| json!({})); - // PHP encodes an empty associative map as []; activity-only published - // workers send this shape. Preserve a map internally for lookups. - let contracts = if contracts.as_array().is_some_and(|values| values.is_empty()) { - json!({}) - } else { - contracts - }; - if !contracts.is_object() { - return Err(refuse( - StatusCode::UNPROCESSABLE_ENTITY, - "invalid_worker_definition", - )); - } + .map(|contracts| { + super::worker_contracts::normalize(contracts, &body["supported_workflow_types"]) + }) + .transpose()?; let token = super::worker_incarnations::token().await?; let mut tx = self.begin().await?; - let previous = Self::query("SELECT worker_id FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") + let previous = Self::query("SELECT worker_id,workflow_command_contracts FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") .bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut *tx).await?; + let contracts = match supplied_contracts { + Some(contracts) => contracts, + None => { + let existing = match &previous { + Some(worker) => { + DB::optional_document_row(worker, "workflow_command_contracts")? + .unwrap_or(Value::Null) + } + None => Value::Null, + }; + super::worker_contracts::normalize(&existing, &body["supported_workflow_types"])? + } + }; if previous.is_some() { // Match PHP's registration boundary: return original workflow leases // to ready without a repair command/history event. Token rotation diff --git a/rust/src/runtime/worker_contracts.rs b/rust/src/runtime/worker_contracts.rs new file mode 100644 index 000000000..871249dc3 --- /dev/null +++ b/rust/src/runtime/worker_contracts.rs @@ -0,0 +1,93 @@ +//! Registration declarations become immutable admission snapshots. A missing +//! workflow entry stays unknown; a valid empty entry declares empty lists. +use super::{Result, refuse}; +use axum::http::StatusCode; +use serde_json::{Value, json}; +use std::collections::BTreeSet; + +fn invalid() -> super::RuntimeError { + refuse( + StatusCode::UNPROCESSABLE_ENTITY, + "invalid_worker_definition", + ) +} + +fn name(value: &Value) -> Result<&str> { + let text = value.as_str().ok_or_else(invalid)?; + let text = text.trim_matches([' ', '\t', '\n', '\r', '\0', '\u{b}']); + if text.is_empty() || text.chars().count() > 255 { + return Err(invalid()); + } + Ok(text) +} + +fn list<'a>(entry: &'a Value, field: &str) -> Result<&'a [Value]> { + match entry.get(field) { + None => Ok(&[]), + Some(Value::Array(values)) => Ok(values), + Some(_) => Err(invalid()), + } +} + +pub(super) fn declaration(entry: &Value) -> Result { + if !entry.is_object() && !entry.as_array().is_some_and(Vec::is_empty) { + return Err(invalid()); + } + let mut result = json!({}); + for (names, handlers) in [ + ("queries", "query_contracts"), + ("signals", "signal_contracts"), + ("updates", "update_contracts"), + ] { + let mut declared = BTreeSet::new(); + for value in list(entry, names)? { + declared.insert(name(value)?.to_owned()); + } + let mut seen = BTreeSet::new(); + let mut contracts = Vec::new(); + for handler in list(entry, handlers)? { + let handler_name = name(&handler["name"])?; + declared.insert(handler_name.to_owned()); + if seen.insert(handler_name.to_owned()) { + contracts.push( + json!({"name":handler_name,"parameters":handler.get("parameters") + .filter(|value| value.is_array()).cloned().unwrap_or_else(|| json!([]))}), + ); + } + } + result[names] = json!(declared); + result[handlers] = json!(contracts); + } + if entry.get("update_validators").is_some() { + let mut validators = BTreeSet::new(); + for value in list(entry, "update_validators")? { + validators.insert(name(value)?.to_owned()); + } + result["update_validators"] = json!(validators); + } + Ok(result) +} + +pub(super) fn normalize(value: &Value, supported: &Value) -> Result { + if value.is_null() || value.as_array().is_some_and(Vec::is_empty) { + return Ok(json!({})); + } + let entries = value.as_object().ok_or_else(invalid)?; + let mut result = json!({}); + for (workflow_type, entry) in entries { + // Validate even an unadvertised declaration before filtering it out. + // Otherwise malformed metadata could rotate an existing registration. + let entry = declaration(entry)?; + let workflow_type = workflow_type.trim_matches([' ', '\t', '\n', '\r', '\0', '\u{b}']); + if !workflow_type.is_empty() + && supported.as_array().is_some_and(|types| { + types + .iter() + .any(|value| value.as_str() == Some(workflow_type)) + }) + { + result[workflow_type] = entry; + } + } + Ok(result) +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 8c2b7ba54..0c1f6a19e 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -49,6 +49,9 @@ mod selected_run_signals; #[path = "execution/worker_incarnations.rs"] mod worker_incarnations; +#[path = "execution/worker_contracts.rs"] +mod worker_contracts; + #[path = "execution/worker_deregistration.rs"] mod worker_deregistration; diff --git a/rust/tests/execution/worker_contracts.rs b/rust/tests/execution/worker_contracts.rs new file mode 100644 index 000000000..9dac8ad68 --- /dev/null +++ b/rust/tests/execution/worker_contracts.rs @@ -0,0 +1,173 @@ +use super::*; + +fn definition() -> Value { + json!({"worker_id":"contract-worker","task_queue":"test","runtime":"python", + "supported_workflow_types":["echo"],"supported_activity_types":[]}) +} + +async fn register_contract(app: &Router, declaration: Option) -> (StatusCode, Value) { + let mut body = definition(); + if let Some(declaration) = declaration { + body["workflow_command_contracts"] = declaration; + } + request(app, "POST", "/api/worker/register", body).await +} + +async fn snapshot(app: &Router, started: &Value) -> Value { + let history = run_history(app, &started["workflow_id"], &started["run_id"]).await; + history + .as_array() + .unwrap() + .iter() + .find(|event| event["event_type"] == "WorkflowStarted") + .unwrap()["payload"] + .clone() +} + +fn assert_empty(payload: &Value) { + for field in [ + "declared_queries", + "declared_signals", + "declared_updates", + "declared_query_contracts", + "declared_signal_contracts", + "declared_update_contracts", + ] { + assert_eq!(payload.get(field), Some(&json!([])), "{field}: {payload}"); + } +} + +#[tokio::test] +async fn explicit_empty_declarations_survive_cold_reads_and_later_registration() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let mut originals = Vec::new(); + for (index, declaration) in [ + json!({}), + json!([]), + json!({"queries":[],"signals":[], + "updates":[],"query_contracts":[],"signal_contracts":[],"update_contracts":[]}), + ] + .into_iter() + .enumerate() + { + let registered = register_contract(&app, Some(json!({"echo":declaration}))).await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let started = start(&app, &format!("empty-contract-{index}")).await; + let before = snapshot(&app, &started).await; + assert_empty(&before); + originals.push((started, before)); + } + assert_eq!( + register_contract( + &app, + Some(json!({"echo":{"query_contracts":[ + {"name":"state","parameters":[]}],"signals":["z","a","z"]}})) + ) + .await + .0, + StatusCode::CREATED + ); + let later = start(&app, "named-contract").await; + let named = snapshot(&app, &later).await; + assert_eq!(named["declared_queries"], json!(["state"])); + assert_eq!(named["declared_signals"], json!(["a", "z"])); + assert_eq!(register_contract(&app, None).await.0, StatusCode::CREATED); + let omitted = start(&app, "reregister-omitted").await; + assert_eq!( + snapshot(&app, &omitted).await["declared_queries"], + json!(["state"]) + ); + drop(app); + drop(runtime); + let cold = router(database.open().await.unwrap()); + for (started, before) in originals { + assert_eq!(snapshot(&cold, &started).await, before); + } + drop(cold); + database.remove().await; +} + +#[tokio::test] +async fn legacy_missing_entries_remain_unknown_instead_of_becoming_empty() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + for (index, declaration) in [ + None, + Some(Value::Null), + Some(json!({})), + Some(json!([])), + Some(json!({"other":{}})), + ] + .into_iter() + .enumerate() + { + let registered = register_contract(&app, declaration).await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let original = start(&app, &format!("legacy-contract-{index}")).await; + let before = snapshot(&app, &original).await; + assert!(before.get("declared_queries").is_none(), "{before}"); + assert_eq!( + register_contract(&app, Some(json!({"echo":{}}))).await.0, + StatusCode::CREATED + ); + assert_eq!(snapshot(&app, &original).await, before); + assert_eq!( + register_contract(&app, Some(json!({}))).await.0, + StatusCode::CREATED + ); + } + drop(app); + database.remove().await; +} + +#[tokio::test] +async fn invalid_declarations_refuse_before_registration_rotation_or_lease_release() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let accepted = register_contract(&app, Some(json!({"echo":{}}))).await; + assert_eq!(accepted.0, StatusCode::CREATED); + let started = start(&app, "contract-lease").await; + let task = poll(&app, "contract-worker", "workflow").await; + let before = snapshot(&app, &started).await; + for declaration in [ + Value::Null, + json!("bad"), + json!({"queries":null}), + json!({"queries":"state"}), + json!({"queries":[17]}), + json!({"signals":[" "]}), + json!({"updates":{"key":"set"}}), + json!({"update_validators":null}), + json!({"query_contracts":"bad"}), + json!({"signal_contracts":[null]}), + json!({"query_contracts":[{}]}), + json!({"update_contracts":[{"name":17}]}), + ] { + let refused = register_contract(&app, Some(json!({"echo":declaration}))).await; + assert_eq!(refused.0, StatusCode::UNPROCESSABLE_ENTITY, "{}", refused.1); + assert_eq!(snapshot(&app, &started).await, before); + } + let completed = finish_task( + &app, + &task, + json!([{"type":"complete_workflow", + "result":envelope(Payload::Long(9007199254740993))}]), + ) + .await; + assert_eq!(completed.0, StatusCode::OK, "{}", completed.1); + assert_eq!(completed.1["recorded"], true); + let retired = request_protocol( + &app, + "POST", + "/api/worker/registrations/contract-worker/deregister", + json!({"registration_token":accepted.1["registration_token"]}), + "1.20", + ) + .await; + assert_eq!(retired.0, StatusCode::OK, "{}", retired.1); + assert_eq!(retired.1["recovered_workflow_task_count"], 0); + drop(app); + database.remove().await; +}