From 362fc1b4778a06ce3598ebf849afb59cf39b2b87 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 18:11:29 +0000 Subject: [PATCH] Serve the search attribute registry natively, as published PHP does GET, POST and DELETE /api/search-attributes list, register and remove a namespace's custom definitions for operators, with PHP's rules: trimmed request strings, Laravel's validation messages, reserved system names and aliases, duplicates that keep their type, a per-namespace limit (DW_MAX_SEARCH_ATTRIBUTES, default 100) and protected system attributes. An empty custom map renders as a JSON array, as PHP renders it. A racing duplicate registration is refused as a duplicate rather than a server error. The search-attributes fixture drives two fresh namespaces through the published SDK and raw requests. Start and upsert still refuse attribute values. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 23 + rust/src/runtime/execution.rs | 9 + rust/src/runtime/http.rs | 37 ++ rust/src/runtime/memo_limits.rs | 2 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/search_attributes.rs | 419 ++++++++++++++++++ rust/src/runtime/store.rs | 3 + rust/tests/execution.rs | 3 + rust/tests/execution/search_attributes.rs | 220 +++++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../search-attribute-contract.mjs | 78 ++++ .../server-parity/search-attribute-probe.php | 64 +++ .../ServerParity/search-attributes.json | 86 ++++ ...erverParitySearchAttributeContractTest.mjs | 72 +++ 16 files changed, 1028 insertions(+), 2 deletions(-) create mode 100644 rust/src/runtime/search_attributes.rs create mode 100644 rust/tests/execution/search_attributes.rs create mode 100644 scripts/conformance/server-parity/search-attribute-contract.mjs create mode 100644 scripts/conformance/server-parity/search-attribute-probe.php create mode 100644 tests/Fixtures/ServerParity/search-attributes.json create mode 100644 tests/Unit/ServerParitySearchAttributeContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 31788a264..e512b6c46 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -24,6 +24,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParitySearchAttributeContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -61,6 +62,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParitySearchAttributeContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1420,7 +1422,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 349f23089..558f5843b 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -152,6 +152,29 @@ support is not evidence of multi-node safety. Newest checkpoints first. Each entry is exact to the source, tree and runs it names, including unsuccessful observations. +October 20, search attribute registry: + +- **Source.** Native serves `GET` and `POST /api/search-attributes` and + `DELETE /api/search-attributes/{name}` for operators, in the default or a + named namespace, as PHP's `SearchAttributeController` does. Its rules are: + - request strings are trimmed; + - Laravel's validation messages and error counts; + - system names and their aliases are reserved; + - a duplicate keeps its type; + - `DW_MAX_SEARCH_ATTRIBUTES` (legacy `WORKFLOW_MAX_SEARCH_ATTRIBUTES`, + default 100) limits definitions per namespace; + - system attributes cannot be removed. +- **Concurrent registrations.** When two registrations of one name race, the + unique key decides the winner. PHP surfaces that as a server error; native + refuses the loser as a duplicate. +- **Not covered.** Native start and upsert still refuse search attribute + values, so a definition is registry state only. PHP's enforcement of + definitions on start, upsert and visibility filters needs its own fixtures. +- **Fixture.** `search-attributes` drives two fresh namespaces through the + published SDK and raw requests. Published PHP 2.5.14, embedded PHP (which + declares it inapplicable) and native passed the contract in a local copy of + the SQLite parity step, and the comparison passed. + October 19, before #478, #476, #494 and #460 merged: - **Open native PRs.** [#478](https://github.com/durable-workflow/server/pull/478) diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 5715137b8..15acf25bb 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -160,8 +160,10 @@ impl Runtime { let queries = super::QueryTaskConfig::from_env()?; let limit = super::memo_limits::configured_limit()?; let history_budget = super::history_budget::Config::from_env()?; + let search_attribute_limit = super::search_attributes::configured_limit()?; let mut store = Store::new(sqlite::open(database).await?); store.memo_size_limit = limit; + store.search_attribute_limit = search_attribute_limit; store.history_budget = history_budget; Self::from_storage(Storage::Sqlite(store), token, external, queries).await } @@ -172,8 +174,10 @@ impl Runtime { let queries = super::QueryTaskConfig::from_env()?; let limit = super::memo_limits::configured_limit()?; let history_budget = super::history_budget::Config::from_env()?; + let search_attribute_limit = super::search_attributes::configured_limit()?; let mut store = Store::new(PostgresStorage::open(options).await?.into_pool()); store.memo_size_limit = limit; + store.search_attribute_limit = search_attribute_limit; store.history_budget = history_budget; Self::from_storage(Storage::Postgres(store), token, external, queries).await } @@ -184,8 +188,10 @@ impl Runtime { let queries = super::QueryTaskConfig::from_env()?; let limit = super::memo_limits::configured_limit()?; let history_budget = super::history_budget::Config::from_env()?; + let search_attribute_limit = super::search_attributes::configured_limit()?; let mut store = Store::new(MySqlStorage::open(options).await?.into_pool()); store.memo_size_limit = limit; + store.search_attribute_limit = search_attribute_limit; store.history_budget = history_budget; Self::from_storage(Storage::MySql(store), token, external, queries).await } @@ -288,6 +294,9 @@ impl Runtime { delegate!(list_namespaces() -> Result); delegate!(describe_namespace(name: &str) -> Result); delegate!(create_namespace(body: Value) -> Result); + delegate!(list_search_attributes() -> Result); + delegate!(create_search_attribute(body: Value) -> Result); + delegate!(delete_search_attribute(name: &str) -> Result); delegate!(set_external_policy(name: &str, body: Value) -> Result); delegate!( upload_external_local( diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index d09fe705f..6f279dab4 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -42,6 +42,14 @@ pub fn router(runtime: Runtime) -> Router { "/api/external-payloads/v1/{reference_id}", get(super::external_transport::fetch), ) + .route( + "/api/search-attributes", + get(list_search_attributes).post(create_search_attribute), + ) + .route( + "/api/search-attributes/{name}", + delete(delete_search_attribute), + ) .route("/api/workflows", get(list_workflows).post(start)) .route("/api/schedules", post(super::schedule_http::create)) .route( @@ -405,6 +413,16 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { if path == "/api/workflows" { return method == "POST"; } + // The registry is namespace-scoped state with no default-namespace selector. + if path == "/api/search-attributes" { + return matches!(method, "GET" | "POST"); + } + if path + .strip_prefix("/api/search-attributes/") + .is_some_and(|name| !name.is_empty() && !name.contains('/')) + { + return method == "DELETE"; + } if let Some(path) = path.strip_prefix("/api/workflows/") { let parts: Vec<_> = path.split('/').collect(); if matches!( @@ -475,6 +493,25 @@ async fn create_namespace( )) } +async fn list_search_attributes(Extension(runtime): Extension) -> Result> { + Ok(Json(runtime.list_search_attributes().await?)) +} +async fn create_search_attribute( + Extension(runtime): Extension, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + Ok(( + StatusCode::CREATED, + Json(runtime.create_search_attribute(body).await?), + )) +} +async fn delete_search_attribute( + Extension(runtime): Extension, + Path(name): Path, +) -> Result> { + Ok(Json(runtime.delete_search_attribute(&name).await?)) +} + async fn admission_error( request: Request, worker: bool, diff --git a/rust/src/runtime/memo_limits.rs b/rust/src/runtime/memo_limits.rs index 5469ba918..37af2ef3a 100644 --- a/rust/src/runtime/memo_limits.rs +++ b/rust/src/runtime/memo_limits.rs @@ -10,7 +10,7 @@ static NUMERIC_PREFIX: LazyLock = LazyLock::new(|| { .expect("static PHP numeric prefix") }); -fn typed_null(value: &str) -> bool { +pub(super) fn typed_null(value: &str) -> bool { matches!(value.to_ascii_lowercase().as_str(), "null" | "(null)") } diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 9f5766b20..f99645da1 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -46,6 +46,7 @@ mod schedule_execution; mod schedule_http; mod schedule_storage; mod schedules; +mod search_attributes; mod schema; mod selection_summary; mod selections; diff --git a/rust/src/runtime/search_attributes.rs b/rust/src/runtime/search_attributes.rs new file mode 100644 index 000000000..2a3bfc46c --- /dev/null +++ b/rust/src/runtime/search_attributes.rs @@ -0,0 +1,419 @@ +//! Custom search attribute definitions per namespace, as PHP's +//! SearchAttributeController registers them. Native start and upsert do not +//! accept search attribute values yet, so a definition is registry state only. +use super::{ + Result, RuntimeError, + backend::Backend, + memo_limits, + store::{Store, now}, +}; +use axum::http::StatusCode; +use serde_json::{Map, Value, json}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; +use std::env; + +/// PHP's `SearchAttributeDefinition::SYSTEM_ATTRIBUTES`. +const SYSTEM_ATTRIBUTES: [(&str, &str); 11] = [ + ("WorkflowType", "keyword"), + ("WorkflowId", "keyword"), + ("RunId", "keyword"), + ("Status", "keyword"), + ("ExecutionStatus", "keyword"), + ("StartTime", "datetime"), + ("ExecutionTime", "datetime"), + ("CloseTime", "datetime"), + ("TaskQueue", "keyword"), + ("BuildId", "keyword"), + ("BuildIds", "keyword"), +]; +const SYSTEM_ALIASES: [&str; 12] = [ + "workflow_type", + "workflow_id", + "wf_id", + "run_id", + "status", + "execution_status", + "start_time", + "execution_time", + "close_time", + "task_queue", + "build_id", + "build_ids", +]; +const ALLOWED_TYPES: [&str; 8] = [ + "keyword", + "string", + "text", + "int", + "double", + "bool", + "datetime", + "keyword_list", +]; +const DEFAULT_LIMIT: i64 = 100; + +/// `server.limits.max_search_attributes`: DW-primary, WORKFLOW-legacy, cast +/// to an integer as Laravel's env() and PHP's (int) do. +pub(super) fn configured_limit() -> Result { + let read = |key| match env::var(key) { + Ok(value) => Ok(Some(value)), + Err(env::VarError::NotPresent) => Ok(None), + Err(env::VarError::NotUnicode(_)) => Err(super::refuse( + StatusCode::BAD_REQUEST, + "invalid_search_attribute_configuration", + )), + }; + let primary = read("DW_MAX_SEARCH_ATTRIBUTES")?.filter(|value| !memo_limits::typed_null(value)); + Ok(match primary.or(read("WORKFLOW_MAX_SEARCH_ATTRIBUTES")?) { + Some(value) => memo_limits::cast_integer(&value), + None => DEFAULT_LIMIT, + }) +} + +fn system_attributes() -> Value { + Value::Object( + SYSTEM_ATTRIBUTES + .iter() + .map(|(name, kind)| ((*name).to_owned(), json!(kind))) + .collect(), + ) +} + +fn reserved(name: &str) -> bool { + SYSTEM_ATTRIBUTES.iter().any(|(system, _)| *system == name) + || SYSTEM_ALIASES.contains(&name.to_ascii_lowercase().as_str()) + || name.starts_with("__") +} + +/// Laravel's TrimStrings and ConvertEmptyStringsToNull run before validation. +fn normalized(value: Option<&Value>) -> Option { + let trimmed = |c: char| c.is_whitespace() || matches!(c, '\u{feff}' | '\u{200b}' | '\u{200e}'); + match value { + Some(Value::String(text)) => { + let text = text.trim_matches(trimmed); + Some(if text.is_empty() { + Value::Null + } else { + json!(text) + }) + } + other => other.cloned(), + } +} + +/// Laravel's `required`: null, an empty string and an empty array fail. +fn present(value: Option<&Value>) -> bool { + match value { + None | Some(Value::Null) => false, + Some(Value::String(text)) => !text.trim().is_empty(), + Some(Value::Array(items)) => !items.is_empty(), + Some(Value::Object(items)) => !items.is_empty(), + Some(_) => true, + } +} + +/// The messages for `required|string|max:128|regex:/^[a-zA-Z][a-zA-Z0-9_]*$/`. +/// A failed `required` stops the remaining rules, as Laravel's implicit rules do. +fn name_messages(value: Option<&Value>) -> Vec<&'static str> { + if !present(value) { + return vec!["The name field is required."]; + } + let mut messages = Vec::new(); + let text = value.and_then(Value::as_str); + if text.is_none() { + messages.push("The name field must be a string."); + } + // Laravel measures strings in characters and arrays in items; the rule + // set has no array rule, so the message is the string one either way. + let size = match value { + Some(Value::String(text)) => text.chars().count(), + Some(Value::Array(items)) => items.len(), + Some(Value::Object(items)) => items.len(), + _ => 0, + }; + if size > 128 { + messages.push("The name field must not be greater than 128 characters."); + } + // A number's string form never starts with a letter, and other values + // fail the regex rule outright. + let matches = text.is_some_and(|text| { + let mut chars = text.chars(); + chars.next().is_some_and(|c| c.is_ascii_alphabetic()) + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_') + }); + if !matches { + messages.push("The name field format is invalid."); + } + messages +} + +/// The messages for `required|string|in:`. +fn type_messages(value: Option<&Value>) -> Vec<&'static str> { + if !present(value) { + return vec!["The type field is required."]; + } + let text = value.and_then(Value::as_str); + let mut messages = Vec::new(); + if text.is_none() { + messages.push("The type field must be a string."); + } + if !text.is_some_and(|text| ALLOWED_TYPES.contains(&text)) { + messages.push("The selected type is invalid."); + } + messages +} + +/// Laravel's 422 ValidationException body for control-plane requests. +fn validation_failure(errors: Vec<(&'static str, Vec<&'static str>)>) -> RuntimeError { + let all: Vec<&str> = errors + .iter() + .flat_map(|(_, messages)| messages.iter().copied()) + .collect(); + let mut message = all[0].to_owned(); + match all.len() - 1 { + 0 => {} + 1 => message.push_str(" (and 1 more error)"), + more => message.push_str(&format!(" (and {more} more errors)")), + } + let errors: Map = errors + .into_iter() + .map(|(field, messages)| (field.to_owned(), json!(messages))) + .collect(); + RuntimeError::Protocol { + status: StatusCode::UNPROCESSABLE_ENTITY, + response: json!({"message":message,"reason":"validation_failed", + "errors":errors.clone(),"validation_errors":errors}), + } +} + +fn conflict(status: StatusCode, response: Value) -> RuntimeError { + RuntimeError::Protocol { status, response } +} + +impl Store +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> &'q str: Encode<'q, DB> + Type, + for<'q> Option<&'q str>: Encode<'q, DB>, + for<'q> Option: Encode<'q, DB>, + for<'q> String: Encode<'q, DB> + Type, + for<'q> i64: Encode<'q, DB> + Type, + for<'r> i64: Decode<'r, DB>, + usize: ColumnIndex, +{ + pub(super) async fn list_search_attributes(&self) -> Result { + let rows = Self::query( + "SELECT name,type FROM search_attribute_definitions WHERE namespace=$1 ORDER BY name", + ) + .bind(&*self.namespace) + .fetch_all(&self.pool) + .await?; + let mut custom = Map::new(); + for row in &rows { + custom.insert(DB::string(row, "name")?, json!(DB::string(row, "type")?)); + } + // PHP renders an empty map as a JSON array. + let custom = if custom.is_empty() { + json!([]) + } else { + Value::Object(custom) + }; + Ok(json!({"system_attributes":system_attributes(),"custom_attributes":custom})) + } + + async fn existing_attribute_conflict(&self, name: &str) -> Result { + let row = Self::query( + "SELECT name,type FROM search_attribute_definitions WHERE namespace=$1 AND name=$2", + ) + .bind(&*self.namespace) + .bind(name) + .fetch_one(&self.pool) + .await?; + Ok(self.already_exists(name, &row)?) + } + + fn already_exists(&self, name: &str, row: &DB::Row) -> Result { + Ok(conflict( + StatusCode::CONFLICT, + json!({"message":format!("A custom search attribute [{name}] already exists in namespace [{}].", self.namespace), + "reason":"attribute_already_exists","name":DB::string(row,"name")?,"type":DB::string(row,"type")?}), + )) + } + + pub(super) async fn create_search_attribute(&self, body: Value) -> Result { + let name = normalized(body.get("name")); + let kind = normalized(body.get("type")); + let errors: Vec<_> = [ + ("name", name_messages(name.as_ref())), + ("type", type_messages(kind.as_ref())), + ] + .into_iter() + .filter(|(_, messages)| !messages.is_empty()) + .collect(); + if !errors.is_empty() { + return Err(validation_failure(errors)); + } + let name = name.as_ref().and_then(Value::as_str).unwrap_or_default(); + let kind = kind.as_ref().and_then(Value::as_str).unwrap_or_default(); + if reserved(name) { + return Err(conflict( + StatusCode::CONFLICT, + json!({"message":format!("The name [{name}] is reserved for system search attributes."), + "reason":"name_reserved"}), + )); + } + let mut tx = self.begin().await?; + if let Some(row) = Self::query( + "SELECT name,type FROM search_attribute_definitions WHERE namespace=$1 AND name=$2", + ) + .bind(&*self.namespace) + .bind(name) + .fetch_optional(&mut *tx) + .await? + { + return Err(self.already_exists(name, &row)?); + } + let count = + Self::scalar("SELECT COUNT(*) FROM search_attribute_definitions WHERE namespace=$1") + .bind(&*self.namespace) + .fetch_one(&mut *tx) + .await?; + if count >= self.search_attribute_limit { + let limit = self.search_attribute_limit; + return Err(conflict( + StatusCode::UNPROCESSABLE_ENTITY, + json!({"message":format!("Namespace [{}] has reached the maximum of {limit} custom search attributes.", self.namespace), + "reason":"search_attribute_limit_reached","limit":limit}), + )); + } + let timestamp = now(); + let inserted = Self::query("INSERT INTO search_attribute_definitions(namespace,name,type,created_at,updated_at) VALUES ($1,$2,$3,$4,$5)") + .bind(&*self.namespace).bind(name).bind(kind) + .bind(DB::bind_time(timestamp)).bind(DB::bind_time(timestamp)) + .execute(&mut *tx) + .await; + match inserted { + Ok(_) => tx.commit().await?, + // A concurrent registration of the same name committed first. + // PHP surfaces the unique-key failure as a server error; refuse + // it the same way as a duplicate found before the insert. + Err(sqlx::Error::Database(error)) if error.is_unique_violation() => { + drop(tx); + return Err(self.existing_attribute_conflict(name).await?); + } + Err(error) => return Err(error.into()), + } + Ok(json!({"name":name,"type":kind,"outcome":"created"})) + } + + pub(super) async fn delete_search_attribute(&self, name: &str) -> Result { + let deleted = + Self::query("DELETE FROM search_attribute_definitions WHERE namespace=$1 AND name=$2") + .bind(&*self.namespace) + .bind(name) + .execute(&self.pool) + .await?; + if DB::affected(deleted) > 0 { + return Ok(json!({"name":name,"outcome":"deleted"})); + } + if SYSTEM_ATTRIBUTES.iter().any(|(system, _)| *system == name) { + return Err(conflict( + StatusCode::CONFLICT, + json!({"message":format!("The system search attribute [{name}] cannot be removed."), + "reason":"system_attribute"}), + )); + } + Err(conflict( + StatusCode::NOT_FOUND, + json!({"message":format!("Custom search attribute [{name}] not found in namespace [{}].", self.namespace), + "reason":"attribute_not_found"}), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn messages(body: Value) -> Option { + let name = normalized(body.get("name")); + let kind = normalized(body.get("type")); + let errors: Vec<_> = [ + ("name", name_messages(name.as_ref())), + ("type", type_messages(kind.as_ref())), + ] + .into_iter() + .filter(|(_, messages)| !messages.is_empty()) + .collect(); + if errors.is_empty() { + return None; + } + match validation_failure(errors) { + RuntimeError::Protocol { response, .. } => Some(response), + _ => unreachable!(), + } + } + + #[test] + fn validation_messages_follow_laravel_rules() { + assert_eq!(messages(json!({"name":"orderTotal","type":"double"})), None); + assert_eq!( + messages(json!({"name":" orderTotal\n","type":" int "})), + None + ); + let missing = messages(json!({})).unwrap(); + assert_eq!( + missing["message"], + "The name field is required. (and 1 more error)" + ); + assert_eq!( + missing["errors"], + json!({"name":["The name field is required."],"type":["The type field is required."]}) + ); + assert_eq!(missing["validation_errors"], missing["errors"]); + assert_eq!( + messages(json!({"name":" ","type":"int"})).unwrap()["errors"], + json!({"name":["The name field is required."]}) + ); + assert_eq!( + messages(json!({"name":[],"type":"int"})).unwrap()["errors"], + json!({"name":["The name field is required."]}) + ); + let numeric = messages(json!({"name":7,"type":5})).unwrap(); + assert_eq!( + numeric["message"], + "The name field must be a string. (and 3 more errors)" + ); + assert_eq!( + numeric["errors"], + json!({ + "name":["The name field must be a string.","The name field format is invalid."], + "type":["The type field must be a string.","The selected type is invalid."]}) + ); + assert_eq!( + messages(json!({"name":"1st","type":"uuid"})).unwrap()["errors"], + json!({"name":["The name field format is invalid."],"type":["The selected type is invalid."]}) + ); + let long = "a".repeat(129); + assert_eq!( + messages(json!({"name":long,"type":"int"})).unwrap()["message"], + "The name field must not be greater than 128 characters." + ); + assert_eq!(messages(json!({"name":"a".repeat(128),"type":"int"})), None); + assert_eq!( + messages(json!({"name":"has-dash","type":"keyword_list"})).unwrap()["message"], + "The name field format is invalid." + ); + } + + #[test] + fn reserved_names_cover_system_attributes_and_aliases() { + for name in ["WorkflowType", "BuildIds", "workflow_ID", "Wf_Id", "status"] { + assert!(reserved(name), "{name}"); + } + for name in ["workflowType", "OrderTotal", "Statuses"] { + assert!(!reserved(name), "{name}"); + } + } +} diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index f944c8c58..113f62d5d 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -25,6 +25,7 @@ pub(super) struct Store { pub(super) memo_codec: Arc, pub(super) memo_size_limit: Option, pub(super) history_budget: super::history_budget::Config, + pub(super) search_attribute_limit: i64, } pub(super) fn id() -> String { Ulid::new().to_string() @@ -66,6 +67,7 @@ where memo_codec: Arc::new(Semaphore::new(1)), memo_size_limit: Some(262_144), history_budget: super::history_budget::Config::default(), + search_attribute_limit: 100, } } pub(super) fn in_namespace(&self, namespace: &str) -> Self { @@ -76,6 +78,7 @@ where memo_codec: self.memo_codec.clone(), memo_size_limit: self.memo_size_limit, history_budget: self.history_budget, + search_attribute_limit: self.search_attribute_limit, } } pub(super) async fn database_live(&self) -> bool { diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index a9097d460..3c72f6634 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -37,6 +37,9 @@ mod role_tokens; #[path = "execution/namespaces.rs"] mod namespaces; +#[path = "execution/search_attributes.rs"] +mod search_attributes; + #[path = "execution/named_timers.rs"] mod named_timers; diff --git a/rust/tests/execution/search_attributes.rs b/rust/tests/execution/search_attributes.rs new file mode 100644 index 000000000..d74b72f0c --- /dev/null +++ b/rust/tests/execution/search_attributes.rs @@ -0,0 +1,220 @@ +use super::*; + +async fn call( + app: &Router, + token: &str, + namespace: &str, + method: &str, + path: &str, + body: Value, +) -> (StatusCode, Value) { + let (status, _, body) = admission::admission_http( + app, + method, + path, + false, + Some(token), + Some("2"), + Some(namespace), + body, + ) + .await; + (status, body) +} + +#[tokio::test] +async fn search_attribute_definitions_follow_the_published_registry_rules() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + namespaces::create(&app, "attributes").await; + let scoped = |method: &'static str, path: &'static str, body: Value| { + let app = app.clone(); + async move { namespaces::scoped(&app, "attributes", method, path, body).await } + }; + + let (status, empty) = scoped("GET", "/api/search-attributes", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{empty}"); + assert_eq!(empty["custom_attributes"], json!([])); + assert_eq!(empty["system_attributes"]["WorkflowType"], "keyword"); + assert_eq!(empty["system_attributes"]["StartTime"], "datetime"); + assert_eq!(empty["system_attributes"].as_object().unwrap().len(), 11); + + let (status, created) = scoped( + "POST", + "/api/search-attributes", + json!({"name":" OrderTotal ","type":"double"}), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{created}"); + assert_eq!( + created, + json!({"name":"OrderTotal","type":"double","outcome":"created"}) + ); + scoped( + "POST", + "/api/search-attributes", + json!({"name":"Customer","type":"keyword"}), + ) + .await; + + let listed = scoped("GET", "/api/search-attributes", Value::Null).await.1; + assert_eq!( + listed["custom_attributes"], + json!({"Customer":"keyword","OrderTotal":"double"}) + ); + // Definitions belong to one namespace. + let default = request(&app, "GET", "/api/search-attributes", Value::Null).await; + assert_eq!(default.1["custom_attributes"], json!([])); + + let (status, duplicate) = scoped( + "POST", + "/api/search-attributes", + json!({"name":"OrderTotal","type":"int"}), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!( + duplicate, + json!({"message":"A custom search attribute [OrderTotal] already exists in namespace [attributes].", + "reason":"attribute_already_exists","name":"OrderTotal","type":"double"}) + ); + for name in ["WorkflowId", "task_queue", "Execution_Status"] { + let (status, reserved) = scoped( + "POST", + "/api/search-attributes", + json!({"name":name,"type":"keyword"}), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{name}"); + assert_eq!(reserved["reason"], "name_reserved"); + assert_eq!( + reserved["message"], + format!("The name [{name}] is reserved for system search attributes.") + ); + } + let (status, invalid) = scoped( + "POST", + "/api/search-attributes", + json!({"name":"1st","type":"uuid"}), + ) + .await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!( + invalid, + json!({"message":"The name field format is invalid. (and 1 more error)","reason":"validation_failed", + "errors":{"name":["The name field format is invalid."],"type":["The selected type is invalid."]}, + "validation_errors":{"name":["The name field format is invalid."],"type":["The selected type is invalid."]}}) + ); + + let (status, system) = + scoped("DELETE", "/api/search-attributes/WorkflowType", Value::Null).await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!( + system, + json!({"message":"The system search attribute [WorkflowType] cannot be removed.","reason":"system_attribute"}) + ); + let (status, missing) = scoped("DELETE", "/api/search-attributes/Absent", Value::Null).await; + assert_eq!(status, StatusCode::NOT_FOUND); + assert_eq!( + missing, + json!({"message":"Custom search attribute [Absent] not found in namespace [attributes].","reason":"attribute_not_found"}) + ); + let (status, deleted) = scoped("DELETE", "/api/search-attributes/Customer", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{deleted}"); + assert_eq!(deleted, json!({"name":"Customer","outcome":"deleted"})); + assert_eq!( + scoped("GET", "/api/search-attributes", Value::Null).await.1["custom_attributes"], + json!({"OrderTotal":"double"}) + ); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_namespace_holds_at_most_the_configured_number_of_definitions() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + namespaces::create(&app, "attribute-limit").await; + for index in 0..100 { + let (status, body) = namespaces::scoped( + &app, + "attribute-limit", + "POST", + "/api/search-attributes", + json!({"name":format!("Attribute{index}"),"type":"int"}), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + } + let (status, refused) = namespaces::scoped( + &app, + "attribute-limit", + "POST", + "/api/search-attributes", + json!({"name":"OneTooMany","type":"int"}), + ) + .await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!( + refused, + json!({"message":"Namespace [attribute-limit] has reached the maximum of 100 custom search attributes.", + "reason":"search_attribute_limit_reached","limit":100}) + ); + // The limit is per namespace. + let (status, _) = request( + &app, + "POST", + "/api/search-attributes", + json!({"name":"OneTooMany","type":"int"}), + ) + .await; + assert_eq!(status, StatusCode::CREATED); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn operators_manage_definitions_and_workers_cannot() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap().with_role_tokens( + Some("worker-token".into()), + Some("operator-token".into()), + Some("admin-token".into()), + ); + let app = router(runtime.clone()); + let body = json!({"name":"Region","type":"keyword"}); + let (status, refused) = call( + &app, + "worker-token", + "default", + "POST", + "/api/search-attributes", + body.clone(), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{refused}"); + let (status, created) = call( + &app, + "operator-token", + "default", + "POST", + "/api/search-attributes", + body, + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{created}"); + let (status, deleted) = call( + &app, + "operator-token", + "default", + "DELETE", + "/api/search-attributes/Region", + Value::Null, + ) + .await; + assert_eq!(status, StatusCode::OK, "{deleted}"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index e60868655..efad78956 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -9,6 +9,7 @@ import {checkVisibility} from './visibility-contract.mjs'; import {checkAdmission} from './admission-contract.mjs'; import {checkNamespaces} from './namespace-contract.mjs'; import {checkWorkerDeregistration} from './worker-deregistration-contract.mjs'; +import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkPatchDeployment, checkPublishedPatchArtifacts, checkPatchPackageObservation} from './patch-deployment-contract.mjs'; import {checkLegacyMarkerArtifacts, checkLegacyMarkerPackageObservation, checkLegacyMarkerDeployment} from './legacy-marker-contract.mjs'; import {checkWaitingHistory} from './waiting-history-contract.mjs'; @@ -128,6 +129,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const admission = fixture.admission ? checkAdmission(fixture, observation, workflowId) : null; const namespaces = fixture.namespace_isolation ? checkNamespaces(fixture, observation, workflowId, checkObservation) : null; const workerDeregistration = fixture.worker_deregistration ? checkWorkerDeregistration(fixture, observation, workflowId) : null; + const searchAttributes = fixture.search_attributes ? checkSearchAttributes(fixture, observation, workflowId) : null; const patchDeployment = fixture.patch_deployment ? (fixture.patch_deployment.legacy_marker_history ? checkLegacyMarkerDeployment(fixture, observation, workflowId) : checkPatchDeployment(fixture, observation, workflowId)) : null; const waitingHistory = fixture.waiting_for_history ? checkWaitingHistory(fixture, observation, workflowId) : null; @@ -665,6 +667,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(admission ? {admission} : {}), ...(namespaces ? {namespace_isolation: namespaces} : {}), ...(workerDeregistration ? {worker_deregistration: workerDeregistration} : {}), + ...(searchAttributes ? {search_attributes: searchAttributes} : {}), ...(patchDeployment ? {patch_deployment: patchDeployment} : {}), ...(waitingHistory ? {waiting_for_history: waitingHistory} : {}), ...(taskRetry ? {workflow_task_retry: taskRetry} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 060d26745..8c0e8b958 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -427,6 +427,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $handle->selectedRunId, $queue, $url, $namespace) : null; $namespaces = isset($fixture['namespace_isolation']) ? finishHttpNamespaces($client, $fixture, $workflowId, $queue, $url) : null; $workerDeregistration = isset($fixture['worker_deregistration']) ? finishHttpWorkerDeregistration($client, $fixture, $workflowId, $queue, $url) : null; + $searchAttributes = isset($fixture['search_attributes']) ? finishHttpSearchAttributes($client, $fixture, $workflowId, $url) : null; $waitingHistory = isset($fixture['waiting_for_history']) ? finishHttpWaitingHistory($client, $fixture, $workflowId, $queue, $url) : null; $taskRetry = isset($fixture['workflow_task_retry']) ? finishHttpTaskRetry($client, $fixture, $workflowId, $queue, $url) : null; if ($scheduleState !== null) { @@ -513,6 +514,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'admission' => $admission, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, + 'search_attributes' => $searchAttributes, 'waiting_for_history' => $waitingHistory, 'workflow_task_retry' => $taskRetry, 'run_id' => $execution->runId, @@ -808,6 +810,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/visibility-probe.php'; require __DIR__.'/admission-probe.php'; require __DIR__.'/namespace-probe.php'; + require __DIR__.'/search-attribute-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; require __DIR__.'/waiting-history-probe.php'; require __DIR__.'/task-retry-probe.php'; @@ -942,6 +945,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['worker_deregistration'])) { $observation['worker_deregistration'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; } + if ($mode === 'embedded' && isset($fixture['search_attributes'])) { + $observation['search_attributes'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_search_attribute_registry']; + } $observation['mode'] = $mode; $observation['typed_output'] = typedValue($observation['output']); if ($mode === 'embedded' && isset($fixture['waiting_for_history'])) { diff --git a/scripts/conformance/server-parity/search-attribute-contract.mjs b/scripts/conformance/server-parity/search-attribute-contract.mjs new file mode 100644 index 000000000..9dc8de9e0 --- /dev/null +++ b/scripts/conformance/server-parity/search-attribute-contract.mjs @@ -0,0 +1,78 @@ +import assert from 'node:assert/strict'; + +// Reviewed custom search attribute registry (SearchAttributeController with +// Laravel's request normalization and validation messages). Embedded +// applications have no HTTP registry, so the projection is the definition alone. +const SYSTEM_ATTRIBUTES = {WorkflowType: 'keyword', WorkflowId: 'keyword', RunId: 'keyword', Status: 'keyword', + ExecutionStatus: 'keyword', StartTime: 'datetime', ExecutionTime: 'datetime', CloseTime: 'datetime', + TaskQueue: 'keyword', BuildId: 'keyword', BuildIds: 'keyword'}; +const REQUIRED = field => `The ${field} field is required.`; +const STRING = field => `The ${field} field must be a string.`; +const FORMAT = 'The name field format is invalid.'; +const TYPE = 'The selected type is invalid.'; +const LONG = 'The name field must not be greater than 128 characters.'; +// The fixture's invalid bodies, in order. +const INVALID_ERRORS = [ + {name: [REQUIRED('name')], type: [REQUIRED('type')]}, + {name: [FORMAT], type: [TYPE]}, + {name: [STRING('name'), FORMAT], type: [STRING('type'), TYPE]}, + {name: [REQUIRED('name')]}, + {name: [LONG]}, + {name: [FORMAT], type: [STRING('type'), TYPE]}, +]; + +function validation(errors) { + const messages = Object.values(errors).flat(); + const more = messages.length - 1; + const message = messages[0] + (more === 0 ? '' : ` (and ${more} more ${more === 1 ? 'error' : 'errors'})`); + return {status: 422, response: {message, reason: 'validation_failed', errors, validation_errors: errors}}; +} + +export function checkSearchAttributes(fixture, observation, workflowId) { + const definition = fixture.search_attributes; + const state = observation.search_attributes; + const projection = {scope: 'search_attributes', limit: definition.limit}; + assert.ok(state && typeof state === 'object', 'actual search attribute observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_search_attribute_registry'}, + 'embedded search attributes explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const [namespace, limitNamespace] = [`${workflowId}-attributes`, `${workflowId}-attribute-limit`].map(name => name.toLowerCase()); + equal([state.applicable, state.namespaces], [true, [namespace, limitNamespace]], 'registry namespaces'); + // PHP renders an empty map as a JSON array. + const listing = custom => ({system_attributes: SYSTEM_ATTRIBUTES, custom_attributes: custom}); + equal(state.sdk_empty, listing([]), 'a new namespace lists only the system attributes'); + equal(state.sdk_created, definition.created.map(([name, type]) => ({name, type, outcome: 'created'})), + 'the published SDK registers definitions'); + equal(state.sdk_listed, listing(Object.fromEntries(definition.created)), 'the published SDK lists them'); + equal(state.trimmed, {status: 201, response: {name: 'Region', type: 'keyword_list', outcome: 'created'}}, + 'request strings are trimmed before validation'); + equal(state.other_namespace, {status: 200, response: listing([])}, 'definitions belong to one namespace'); + const [name, type] = definition.created.find(([created]) => created === definition.duplicate.name); + equal(state.duplicate, {status: 409, response: { + message: `A custom search attribute [${name}] already exists in namespace [${namespace}].`, + reason: 'attribute_already_exists', name, type}}, 'a duplicate keeps the original definition'); + equal(state.reserved, definition.reserved.map(reserved => ({status: 409, response: { + message: `The name [${reserved}] is reserved for system search attributes.`, reason: 'name_reserved'}})), + 'system names and their aliases are reserved'); + equal(definition.invalid.length, INVALID_ERRORS.length, 'every invalid body has reviewed messages'); + equal(state.invalid, INVALID_ERRORS.map(validation), 'Laravel validation messages'); + equal(state.delete_system, {status: 409, response: { + message: 'The system search attribute [WorkflowType] cannot be removed.', reason: 'system_attribute'}}, + 'system attributes cannot be removed'); + equal(state.delete_missing, {status: 404, response: { + message: `Custom search attribute [Absent] not found in namespace [${namespace}].`, reason: 'attribute_not_found'}}, + 'an absent definition is not found'); + const [removed] = definition.created[0]; + equal(state.sdk_deleted, {name: removed, outcome: 'deleted'}, 'the published SDK removes a definition'); + const remaining = Object.fromEntries([...definition.created.slice(1), ['Region', 'keyword_list']]); + equal(state.listed_after_delete, {status: 200, response: listing(remaining)}, 'the removal is listed'); + assert.ok(!('limit_fill_refusal' in state), 'definitions up to the limit are accepted'); + equal(state.over_limit, {status: 422, response: { + message: `Namespace [${limitNamespace}] has reached the maximum of ${definition.limit} custom search attributes.`, + reason: 'search_attribute_limit_reached', limit: definition.limit}}, 'the namespace limit is enforced'); + equal(state.limit_count, definition.limit, 'the refused definition was not stored'); + return projection; +} diff --git a/scripts/conformance/server-parity/search-attribute-probe.php b/scripts/conformance/server-parity/search-attribute-probe.php new file mode 100644 index 000000000..aeb51e554 --- /dev/null +++ b/scripts/conformance/server-parity/search-attribute-probe.php @@ -0,0 +1,64 @@ +createNamespace($name, 'search attributes λ', 30); + } + $token = getenv('DW_PARITY_TOKEN') ?: null; + $client = parityClient($url, namespace: $names[0], token: $token); + $http = new SearchAttributeHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 10]); + $raw = static function (string $namespace, string $method, string $path, array|object|null $body = null) use ($http, $token): array { + $response = $http->request($method, $path, ['headers' => array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', + ]), ...($body === null ? [] : ['json' => $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $state = ['applicable' => true, 'namespaces' => $names]; + $state['sdk_empty'] = $client->listSearchAttributes()->raw; + foreach ($definition['created'] as [$name, $type]) { + $state['sdk_created'][] = $client->createSearchAttribute($name, $type)->raw; + } + $state['sdk_listed'] = $client->listSearchAttributes()->raw; + // Laravel trims request strings before validation. + $state['trimmed'] = $raw($names[0], 'POST', 'search-attributes', $definition['untrimmed']); + $state['other_namespace'] = $raw($names[1], 'GET', 'search-attributes'); + $state['duplicate'] = $raw($names[0], 'POST', 'search-attributes', $definition['duplicate']); + foreach ($definition['reserved'] as $name) { + $state['reserved'][] = $raw($names[0], 'POST', 'search-attributes', ['name' => $name, 'type' => 'keyword']); + } + foreach ($definition['invalid'] as $body) { + $state['invalid'][] = $raw($names[0], 'POST', 'search-attributes', $body === [] ? (object) [] : $body); + } + $state['delete_system'] = $raw($names[0], 'DELETE', 'search-attributes/WorkflowType'); + $state['delete_missing'] = $raw($names[0], 'DELETE', 'search-attributes/Absent'); + $state['sdk_deleted'] = $client->deleteSearchAttribute($definition['created'][0][0])->raw; + $state['listed_after_delete'] = $raw($names[0], 'GET', 'search-attributes'); + for ($index = 0; $index < $definition['limit']; $index++) { + $created = $raw($names[1], 'POST', 'search-attributes', ['name' => 'Attribute'.$index, 'type' => 'int']); + if ($created['status'] !== 201) { + $state['limit_fill_refusal'] = ['index' => $index, ...$created]; + break; + } + } + $state['over_limit'] = $raw($names[1], 'POST', 'search-attributes', ['name' => 'OneTooMany', 'type' => 'int']); + $state['limit_count'] = count($raw($names[1], 'GET', 'search-attributes')['response']['custom_attributes']); + + return $state; +} diff --git a/tests/Fixtures/ServerParity/search-attributes.json b/tests/Fixtures/ServerParity/search-attributes.json new file mode 100644 index 000000000..798895148 --- /dev/null +++ b/tests/Fixtures/ServerParity/search-attributes.json @@ -0,0 +1,86 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "search-attributes", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": { + "message": "search attributes λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "search attributes λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "search_attributes": { + "created": [ + [ + "Customer", + "keyword" + ], + [ + "OrderTotal", + "double" + ] + ], + "untrimmed": { + "name": " Region\n", + "type": " keyword_list " + }, + "duplicate": { + "name": "OrderTotal", + "type": "int" + }, + "reserved": [ + "WorkflowId", + "task_queue", + "Execution_Status" + ], + "invalid": [ + {}, + { + "name": "1st", + "type": "uuid" + }, + { + "name": 7, + "type": 5 + }, + { + "name": " ", + "type": "int" + }, + { + "name": "Aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "type": "int" + }, + { + "name": "has-dash", + "type": [ + "keyword" + ] + } + ], + "limit": 100 + }, + "contract": "Each namespace has its own custom search attribute registry. The unchanged published SDK lists, registers and removes definitions: a new namespace lists the eleven system attributes and renders an empty custom map as a JSON array. Request strings are trimmed before validation. Laravel validation refuses missing, non-string, malformed, over-long and unknown-type fields with 422 validation_failed and the published messages and error counts. A duplicate name is refused with 409 attribute_already_exists and keeps its original type; system names and their snake_case aliases, in any case, are refused with 409 name_reserved. A system attribute cannot be removed (409 system_attribute) and an absent one is 404 attribute_not_found. A namespace accepts 100 definitions and refuses the next with 422 search_attribute_limit_reached without storing it. Search attribute values on start and upsert, and visibility filters on them, need separate fixtures. Embedded applications have no HTTP registry." +} diff --git a/tests/Unit/ServerParitySearchAttributeContractTest.mjs b/tests/Unit/ServerParitySearchAttributeContractTest.mjs new file mode 100644 index 000000000..a2b60950a --- /dev/null +++ b/tests/Unit/ServerParitySearchAttributeContractTest.mjs @@ -0,0 +1,72 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkSearchAttributes} from '../../scripts/conformance/server-parity/search-attribute-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/search-attributes.json', import.meta.url))); +const system = {WorkflowType: 'keyword', WorkflowId: 'keyword', RunId: 'keyword', Status: 'keyword', + ExecutionStatus: 'keyword', StartTime: 'datetime', ExecutionTime: 'datetime', CloseTime: 'datetime', + TaskQueue: 'keyword', BuildId: 'keyword', BuildIds: 'keyword'}; +const invalid = errors => { + const messages = Object.values(errors).flat(); + const more = messages.length - 1; + return {status: 422, response: {message: messages[0] + (more ? ` (and ${more} more error${more === 1 ? '' : 's'})` : ''), + reason: 'validation_failed', errors, validation_errors: errors}}; +}; +function model() { + const listing = custom => ({system_attributes: system, custom_attributes: custom}); + const format = 'The name field format is invalid.'; + return structuredClone({mode: 'http', search_attributes: {applicable: true, + namespaces: ['test-attributes', 'test-attribute-limit'], sdk_empty: listing([]), + sdk_created: [{name: 'Customer', type: 'keyword', outcome: 'created'}, {name: 'OrderTotal', type: 'double', outcome: 'created'}], + sdk_listed: listing({Customer: 'keyword', OrderTotal: 'double'}), + trimmed: {status: 201, response: {name: 'Region', type: 'keyword_list', outcome: 'created'}}, + other_namespace: {status: 200, response: listing([])}, + duplicate: {status: 409, response: {message: 'A custom search attribute [OrderTotal] already exists in namespace [test-attributes].', + reason: 'attribute_already_exists', name: 'OrderTotal', type: 'double'}}, + reserved: fixture.search_attributes.reserved.map(name => ({status: 409, response: { + message: `The name [${name}] is reserved for system search attributes.`, reason: 'name_reserved'}})), + invalid: [ + invalid({name: ['The name field is required.'], type: ['The type field is required.']}), + invalid({name: [format], type: ['The selected type is invalid.']}), + invalid({name: ['The name field must be a string.', format], type: ['The type field must be a string.', 'The selected type is invalid.']}), + invalid({name: ['The name field is required.']}), + invalid({name: ['The name field must not be greater than 128 characters.']}), + invalid({name: [format], type: ['The type field must be a string.', 'The selected type is invalid.']}), + ], + delete_system: {status: 409, response: {message: 'The system search attribute [WorkflowType] cannot be removed.', reason: 'system_attribute'}}, + delete_missing: {status: 404, response: {message: 'Custom search attribute [Absent] not found in namespace [test-attributes].', + reason: 'attribute_not_found'}}, + sdk_deleted: {name: 'Customer', outcome: 'deleted'}, + listed_after_delete: {status: 200, response: listing({OrderTotal: 'double', Region: 'keyword_list'})}, + over_limit: {status: 422, response: {message: 'Namespace [test-attribute-limit] has reached the maximum of 100 custom search attributes.', + reason: 'search_attribute_limit_reached', limit: 100}}, + limit_count: 100}}); +} +test('published registry behavior passes', () => + assert.deepStrictEqual(checkSearchAttributes(fixture, model(), 'test'), {scope: 'search_attributes', limit: 100})); +test('embedded explicitly declares the HTTP registry inapplicable', () => checkSearchAttributes(fixture, + {mode: 'embedded', search_attributes: {applicable: false, reason: 'embedded_has_no_http_search_attribute_registry'}}, 'test')); +for (const [label, mutate] of [ + ['an empty map rendered as an object', s => s.sdk_empty.custom_attributes = {}], + ['a missing system attribute', s => delete s.sdk_listed.system_attributes.BuildIds], + ['untrimmed names stored', s => s.trimmed.response.name = ' Region\n'], + ['definitions shared across namespaces', s => s.other_namespace.response.custom_attributes = {Customer: 'keyword'}], + ['a duplicate that replaces the type', s => s.duplicate.response.type = 'int'], + ['an alias accepted', s => s.reserved[1] = {status: 201, response: {name: 'task_queue', type: 'keyword', outcome: 'created'}}], + ['a validation message changed', s => s.invalid[2].response.errors.name.pop()], + ['the summary miscounts errors', s => s.invalid[0].response.message = 'The name field is required.'], + ['validation errors without the legacy field', s => delete s.invalid[1].response.validation_errors], + ['a system attribute removed', s => s.delete_system = {status: 200, response: {name: 'WorkflowType', outcome: 'deleted'}}], + ['an absent definition removed', s => s.delete_missing.status = 200], + ['a removal not listed', s => s.listed_after_delete.response.custom_attributes.Customer = 'keyword'], + ['the limit not enforced', s => s.over_limit = {status: 201, response: {name: 'OneTooMany', type: 'int', outcome: 'created'}}], + ['a refusal before the limit', s => s.limit_fill_refusal = {index: 99, status: 422}], + ['the refused definition stored', s => s.limit_count = 101], +]) { + test(`rejects ${label}`, () => { + const observation = model(); + mutate(observation.search_attributes); + assert.throws(() => checkSearchAttributes(fixture, observation, 'test')); + }); +}