diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 07514d356..5457019d6 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -42,6 +42,7 @@ on: - tests/Unit/ServerParityWorkerReplacementContractTest.mjs - tests/Unit/ServerParityOperatorMetricsContractTest.mjs - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs + - tests/Unit/ServerParityBridgeAdaptersContractTest.mjs - tests/Unit/ServerParityNexusOperationsContractTest.mjs - tests/Unit/ServerParityRecordLocalActivityContractTest.mjs - tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs @@ -1805,7 +1806,7 @@ jobs: bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \ 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/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.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 tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.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 tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.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/app/Http/Controllers/Api/BridgeAdapterController.php b/app/Http/Controllers/Api/BridgeAdapterController.php index 40393de06..f8a8bd0dd 100644 --- a/app/Http/Controllers/Api/BridgeAdapterController.php +++ b/app/Http/Controllers/Api/BridgeAdapterController.php @@ -21,6 +21,13 @@ class BridgeAdapterController { + /** + * Bridge commands' identities own the `dw-bridge:` request id namespace + * (Server #777), so a caller's signal or update request id may not use + * it. MySQL compares request ids without regard to case. + */ + public const UNRESERVED_REQUEST_ID = 'not_regex:/^\s*dw-bridge:/i'; + public function __construct( private readonly WorkflowStartService $workflowStartService, private readonly WorkflowControlPlane $workflowControlPlane, diff --git a/app/Http/Controllers/Api/WorkflowController.php b/app/Http/Controllers/Api/WorkflowController.php index 12022caf8..9511c8ab2 100644 --- a/app/Http/Controllers/Api/WorkflowController.php +++ b/app/Http/Controllers/Api/WorkflowController.php @@ -537,7 +537,7 @@ public function signal(Request $request, string $workflowId, string $signalName) $validated = $request->validate([ 'input' => ['nullable', 'array'], - 'request_id' => ['nullable', 'string', 'max:255'], + 'request_id' => ['nullable', 'string', 'max:255', BridgeAdapterController::UNRESERVED_REQUEST_ID], 'dry_run' => ['nullable', 'boolean'], ]); @@ -749,7 +749,7 @@ public function update(Request $request, string $workflowId, string $updateName) $validated = $request->validate([ 'input' => ['nullable', 'array'], - 'request_id' => ['nullable', 'string', 'max:255'], + 'request_id' => ['nullable', 'string', 'max:255', BridgeAdapterController::UNRESERVED_REQUEST_ID], 'wait_for' => ['nullable', 'string', 'in:accepted,completed'], 'wait_timeout_seconds' => ['nullable', 'integer', 'min:1', 'max:'.ControlPlaneRequestContract::UPDATE_WAIT_TIMEOUT_SECONDS_MAX], ]); diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 755b075e7..9fc053771 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -82,13 +82,13 @@ Updated October 24, 2026. ([#737](https://github.com/durable-workflow/server/issues/737)). No expectation is edited to make native pass. -- **Parity surface.** Native serves 145 of the 175 API route methods in +- **Parity surface.** Native serves 146 of the 175 API route methods in published 2.11.1's route table, with message streams, update validation, activity cancellation acknowledgement, worker sessions, workflow-task - cancellation scopes, caller-indexed Nexus operation reads, and embedded v2 - and Waterline v1 workflow import. An execution test sends each one through - the native router, and the ones - still answering `501 rust_capability_not_implemented` must match + cancellation scopes, caller-indexed Nexus operation reads, bridge adapter + webhooks, and embedded v2 and Waterline v1 workflow import. An execution + test sends each one through the native router, and the ones still + answering `501 rust_capability_not_implemented` must match [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt). Unlike published PHP, native's embedded v2 import refuses a bundle whose rows reference rows outside it, links only runs of one namespace, and reads @@ -169,7 +169,23 @@ any other unknown path or method answers PHP's own 404 or 405 `unsupported_workflow_command`. When it does, an operation bound to a workflow update must refuse a `wait_timeout_seconds` above the update ceiling of 60 before it runs, as PHP does (Server #775).) - - Bridge adapters. + - (Bridge adapter webhooks are served: `POST /api/bridge-adapters/webhook/{adapter}`, + with the `bridge-adapters` fixture. Native admits bridge updates as the + regular update route does; published 2.11.1 skips that admission, Server #747. + Four further differences from published 2.11.1, all tracked in Server #777: + - A start for an id that another namespace holds is refused as unavailable, + naming neither the type nor the namespace. The draining gate and the + instance reservation are decided under the transition lock, before + anything is written. + - The adapter name is at most 64 bytes. + - Signals and updates are deduplicated inside their own transaction under + the transition lock. Each one is recorded with a fixed-length + `dw-bridge:v1:` identity as its request id, and the commands PHP + recorded before takeover are matched by their metadata. A command + under an identity, accepted or rejected, that records another tuple is + refused `409 bridge_identity_conflict`. + - The regular signal and update routes refuse caller request ids that + start with `dw-bridge:`.) - (Caller-indexed Nexus operation reads are served, with the `nexus-operations` fixture. Native writes no service calls and takeover refuses a source that holds any, so every list is PHP's empty answer; an diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 28e8fb869..1229d146c 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -82,6 +82,12 @@ pub(super) trait Backend: Database { const SEARCH_ATTRIBUTE_DELETE: &str = "DELETE FROM workflow_search_attributes WHERE workflow_run_id=$1 AND \"key\"=$2"; const SEARCH_ATTRIBUTE_KEY_ROWS: &str = "SELECT a.workflow_run_id,a.\"key\" AS attribute_key,a.type,a.value_string,a.value_keyword,a.value_int,a.value_float,CASE WHEN a.value_bool IS NULL THEN NULL WHEN a.value_bool THEN 1 ELSE 0 END AS value_bool,a.value_datetime,a.value_keyword_list FROM workflow_search_attributes a JOIN workflow_runs r ON r.id=a.workflow_run_id WHERE r.namespace=$1 AND a.\"key\"=$2"; const SEARCH_ATTRIBUTE_INHERIT: &str = "INSERT INTO workflow_search_attributes(workflow_run_id,workflow_instance_id,\"key\",type,value_string,value_keyword,value_keyword_list,value_int,value_float,value_bool,value_datetime,upserted_at_sequence,inherited_from_parent,created_at,updated_at) SELECT $1,$2,\"key\",type,value_string,value_keyword,value_keyword_list,value_int,value_float,value_bool,value_datetime,1,true,$3,$4 FROM workflow_search_attributes WHERE workflow_run_id=$5"; + // `duplicateBridgeCommand`: the latest accepted command PHP recorded for + // this adapter, action, key and target name ($6 names the target field), + // filtered in the database as PHP's JSON-path query is. PHP's bridge + // commands carry no identity: no request id for a signal, the raw key + // ($8) for an update. + const BRIDGE_COMMAND: &str = "SELECT workflow_run_id,resolved_workflow_run_id FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND status='accepted' AND context->'server'->'metadata'->>'adapter'=$3 AND context->'server'->'metadata'->>'action'=$4 AND context->'server'->'metadata'->>'idempotency_key'=$5 AND context->'server'->'metadata'->>$6=$7 AND (request_id IS NULL OR request_id=$8) ORDER BY accepted_at DESC,command_sequence DESC LIMIT 1"; fn statement(sql: &'static str) -> Cow<'static, str> { Cow::Borrowed(sql) } @@ -191,6 +197,7 @@ WHERE r.status IN ('pending','running','waiting') ORDER BY t.available_order,t.id LIMIT 100 "#; const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND json_extract(payload,'$.selection_group_id')=$2 ORDER BY sequence LIMIT 1"; + const BRIDGE_COMMAND: &str = "SELECT workflow_run_id,resolved_workflow_run_id FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND status='accepted' AND json_extract(context,'$.server.metadata.adapter')=$3 AND json_extract(context,'$.server.metadata.action')=$4 AND json_extract(context,'$.server.metadata.idempotency_key')=$5 AND json_extract(context,'$.server.metadata.'||$6)=$7 AND (request_id IS NULL OR request_id=$8) ORDER BY accepted_at DESC,command_sequence DESC LIMIT 1"; const REPAIR_COUNTER_MAX: i64 = i64::MAX; const RANGE_REPAIR_TASK_SQL: &str = "SELECT * FROM workflow_tasks WHERE workflow_run_id=$1 AND namespace=$2 AND task_type='workflow' AND status='failed' AND json_extract(payload,'$.replay_blocked')=1 ORDER BY CASE WHEN available_at IS NULL THEN 1 ELSE 0 END,available_at,created_at,id LIMIT 1"; const ACTIVITY_OPEN_TASK_SQL: &str = "SELECT * FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='activity' AND status IN ('ready','leased') AND json_extract(payload,'$.activity_execution_id')=$2 ORDER BY id LIMIT 1"; @@ -464,6 +471,7 @@ impl Backend for MySql { const MEMO_INHERIT_INSERT: &str = "INSERT INTO workflow_memos(workflow_run_id,workflow_instance_id,`key`,value,upserted_at_sequence,inherited_from_parent,created_at,updated_at,portable_value,portable_value_sequence) VALUES ($1,$2,$3,$4,1,true,$5,$6,$7,1)"; const MEMO_READ: &str = "SELECT `key`,value,portable_value,portable_value_sequence,upserted_at_sequence FROM workflow_memos WHERE workflow_run_id=$1 ORDER BY `key`,id LIMIT 101"; const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=? AND event_type='SelectionResolved' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.selection_group_id'))=? ORDER BY sequence LIMIT 1"; + const BRIDGE_COMMAND: &str = "SELECT workflow_run_id,resolved_workflow_run_id FROM workflow_commands WHERE workflow_instance_id=? AND command_type=? AND status='accepted' AND BINARY JSON_UNQUOTE(JSON_EXTRACT(context,'$.server.metadata.adapter'))=BINARY ? AND BINARY JSON_UNQUOTE(JSON_EXTRACT(context,'$.server.metadata.action'))=BINARY ? AND BINARY JSON_UNQUOTE(JSON_EXTRACT(context,'$.server.metadata.idempotency_key'))=BINARY ? AND BINARY JSON_UNQUOTE(JSON_EXTRACT(context,CONCAT('$.server.metadata.',?)))=BINARY ? AND (request_id IS NULL OR BINARY request_id=BINARY ?) ORDER BY accepted_at DESC,command_sequence DESC LIMIT 1"; const REPAIR_COUNTER_MAX: i64 = u32::MAX as i64; const RANGE_REPAIR_TASK_SQL: &str = "SELECT * FROM workflow_tasks WHERE workflow_run_id=? AND namespace=? AND task_type='workflow' AND status='failed' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.replay_blocked'))='true' ORDER BY CASE WHEN available_at IS NULL THEN 1 ELSE 0 END,available_at,created_at,id LIMIT 1"; const ACTIVITY_OPEN_TASK_SQL: &str = "SELECT * FROM workflow_tasks WHERE workflow_run_id=? AND task_type='activity' AND status IN ('ready','leased') AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.activity_execution_id'))=? ORDER BY id LIMIT 1"; @@ -509,6 +517,7 @@ impl Backend for MySql { || sql == Self::SIGNAL_RECEIVED_SEQUENCE_SQL || sql == Self::RANGE_REPAIR_TASK_SQL || sql == Self::SELECTION_RESOLUTION_SQL + || sql == Self::BRIDGE_COMMAND || sql == Self::ACTIVITY_OPEN_TASK_SQL || sql == Self::LOCAL_ACTIVITY_EXECUTIONS || sql == Self::LOCAL_ACTIVITY_ATTEMPTS diff --git a/rust/src/runtime/bridge_adapters.rs b/rust/src/runtime/bridge_adapters.rs new file mode 100644 index 000000000..31001ead1 --- /dev/null +++ b/rust/src/runtime/bridge_adapters.rs @@ -0,0 +1,512 @@ +//! Bridge adapter webhooks, as published Server 2.11.1 serves them +//! (`BridgeAdapterController`): an external system starts, signals or +//! updates a workflow through one idempotent event, and every answer is a +//! `durable-workflow.v2.bridge-adapter-outcome.contract` envelope. +//! +//! Published defects not copied: +//! - Server #747: PHP's bridge update skips external update admission, so an +//! update to a workflow served by an SDK worker is always refused +//! `workflow_definition_unavailable`. Native admits it as the regular +//! update route does. +//! - A start for an instance id another namespace holds is refused without +//! naming that instance's type or namespace. PHP names the reserved type. +//! The draining gate and the instance reservation are decided under the +//! transition lock, before anything is written; PHP's gate reads outside +//! its transaction. +//! - The adapter name is at most 64 bytes, so a derived workflow id fits the +//! 128-byte workflow id contract. +//! - Signals and updates are deduplicated atomically (Server #777). A +//! fixed-length bridge identity, derived from the adapter, action, raw key +//! and target, is the command's `request_id`. The duplicate check runs +//! inside the command's transaction, under the transition lock: the +//! identity first, then the command metadata PHP recorded before takeover. +//! The unique (instance, command type, request id) index backs it, and +//! callers cannot send a `dw-bridge:` request id. Any command under an +//! identity, accepted or rejected, must record the same tuple, or the +//! bridge answers 409 `bridge_identity_conflict`. PHP checks once, outside +//! its transaction, and uses the raw key as an update's request id. +use super::command_context::CommandRequest; +use super::execution::Runtime; +use super::laravel_validation::{Rule, rule_errors}; +use super::transaction::Transaction; +use super::{Result, RuntimeError}; +use axum::http::StatusCode; +use serde_json::{Map, Value, json}; +use sha2::{Digest, Sha256}; + +const SCHEMA: &str = "durable-workflow.v2.bridge-adapter-outcome.contract"; + +const EVENT_RULES: &[(&str, &[Rule])] = &[ + ("action", &[Rule::Required, Rule::String]), + ("idempotency_key", &[Rule::Required, Rule::String, Rule::Max(255)]), + ("target", &[Rule::Required, Rule::Array]), + ("input", &[Rule::Nullable, Rule::Array]), + ("correlation", &[Rule::Nullable, Rule::Array]), +]; + +const START_RULES: &[(&str, &[Rule])] = &[ + ("workflow_type", &[Rule::Required, Rule::String, Rule::Max(255)]), + ("workflow_id", &[Rule::Nullable, Rule::String, Rule::Max(128), Rule::Pattern(routing_name)]), + ("task_queue", &[Rule::Nullable, Rule::String, Rule::Max(255)]), + ("business_key", &[Rule::Nullable, Rule::String, Rule::Max(255)]), + ("duplicate_policy", &[Rule::Nullable, Rule::String, Rule::In(&["reject_duplicate", "use_existing"])]), +]; + +const SIGNAL_RULES: &[(&str, &[Rule])] = &[ + ("workflow_id", &[Rule::Required, Rule::String, Rule::Max(128), Rule::Pattern(routing_name)]), + ("signal_name", &[Rule::Required, Rule::String, Rule::Max(255)]), +]; + +const UPDATE_RULES: &[(&str, &[Rule])] = &[ + ("workflow_id", &[Rule::Required, Rule::String, Rule::Max(128), Rule::Pattern(routing_name)]), + ("update_name", &[Rule::Required, Rule::String, Rule::Max(255)]), + ("wait_for", &[Rule::Nullable, Rule::String, Rule::In(&["accepted", "completed"])]), + ("wait_timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(1), + Rule::Max(super::command_requests::UPDATE_WAIT_TIMEOUT_SECONDS_MAX)]), +]; + +/// The longest adapter name: `bridge--<32 hex>` then fits the +/// 128-byte workflow id contract. +const ADAPTER_MAX_BYTES: usize = 64; + +/// `/^[a-zA-Z0-9._:-]+$/`, for adapter names and workflow ids. +fn routing_name(text: &str) -> bool { + !text.is_empty() && text.bytes().all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-')) +} + +fn no_custom(_: &str, _: Rule) -> Option<&'static str> { + None +} + +/// The rule failures for `body`, as Laravel's `errors()->toArray()`, or +/// `None` when every rule passes. +fn errors(body: &Value, rules: &[(&str, &[Rule])]) -> Option { + let errors: Map = rule_errors(body, rules, no_custom) + .into_iter() + .filter(|(_, messages)| !messages.is_empty()) + .map(|(field, messages)| (field, json!(messages))) + .collect(); + (!errors.is_empty()).then_some(Value::Object(errors)) +} + +/// `outcome`: the contract fields first, then the payload without nulls. +fn envelope(payload: Value) -> Value { + let mut body = Map::new(); + body.insert("schema".into(), json!(SCHEMA)); + body.insert("version".into(), json!(1)); + if let Value::Object(fields) = payload { + body.extend(fields.into_iter().filter(|(_, value)| !value.is_null())); + } + Value::Object(body) +} + +fn answer(status: StatusCode, payload: Value) -> Result<(StatusCode, Value)> { + Ok((status, envelope(payload))) +} + +/// `rejected`: a 422 refusal with `extra` after the event fields. +fn rejected(adapter: &str, action: Option<&str>, key: Option<&str>, reason: &str, extra: Value) -> Result<(StatusCode, Value)> { + let mut payload = json!({"adapter":adapter,"action":action,"accepted":false,"outcome":"rejected","reason":reason,"idempotency_key":key}); + if let (Some(fields), Value::Object(extra)) = (payload.as_object_mut(), extra) { + fields.extend(extra); + } + answer(StatusCode::UNPROCESSABLE_ENTITY, payload) +} + +/// `redactedTarget`: only the routing fields, without nulls or empty strings. +fn redacted(target: &Value) -> Value { + let mut fields = Map::new(); + for name in ["workflow_id", "workflow_type", "signal_name", "update_name", "task_queue", "business_key"] { + match target.get(name) { + None | Some(Value::Null) => {} + Some(Value::String(text)) if text.is_empty() => {} + Some(value) => { + fields.insert(name.into(), value.clone()); + } + } + } + Value::Object(fields) +} + +/// `$target + ['workflow_id' => $id]`: the target keeps its own key. +fn with_workflow_id(target: &Value, workflow_id: &str) -> Value { + let mut target = target.clone(); + if let Some(fields) = target.as_object_mut() { + fields.entry("workflow_id").or_insert_with(|| json!(workflow_id)); + } + target +} + +/// `arrayValue`: the field when it is an array, else an empty one. +fn array_value(body: &Value, field: &str) -> Value { + match body.get(field) { + Some(value @ (Value::Object(_) | Value::Array(_))) => value.clone(), + _ => json!([]), + } +} + +fn empty(value: &Value) -> bool { + match value { + Value::Object(fields) => fields.is_empty(), + Value::Array(items) => items.is_empty(), + _ => true, + } +} + +fn hex(digest: &[u8]) -> String { + digest.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// `workflowIdFor`: `bridge--`. A valid +/// adapter name is its own slug. +fn derived_workflow_id(adapter: &str, key: &str) -> String { + format!("bridge-{adapter}-{}", &hex(&Sha256::digest(key.as_bytes()))[..32]) +} + +/// One bridge command: its identity and the metadata it is recorded with. +pub(crate) struct BridgeKey<'a> { + /// `dw-bridge:v1:` and sha256 over the length-prefixed adapter, action, + /// raw key, target field and target name: fixed length, and distinct for + /// any other tuple. It is the command's `request_id`. + pub(crate) identity: String, + pub(crate) adapter: &'a str, + pub(crate) action: &'a str, + pub(crate) key: &'a str, + pub(crate) field: &'static str, + pub(crate) name: &'a str, +} + +impl<'a> BridgeKey<'a> { + fn new(adapter: &'a str, action: &'a str, key: &'a str, field: &'static str, name: &'a str) -> Self { + let mut digest = Sha256::new(); + for part in [adapter, action, key, field, name] { + digest.update((part.len() as u64).to_be_bytes()); + digest.update(part.as_bytes()); + } + let identity = format!("dw-bridge:v1:{}", hex(&digest.finalize())); + Self { identity, adapter, action, key, field, name } + } + + /// Whether a command's context names this adapter, action, key and + /// target. + fn recorded_in(&self, context: &Value) -> bool { + let metadata = &context["server"]["metadata"]; + metadata["adapter"] == self.adapter && metadata["action"] == self.action + && metadata["idempotency_key"] == self.key && metadata[self.field] == self.name + } + + /// Any command found under this identity, accepted or rejected, must + /// record this exact tuple; anything else fails closed before its + /// receipt is answered. + pub(crate) fn verify(&self, context: Option<&Value>) -> Result<()> { + if context.is_some_and(|context| self.recorded_in(context)) { Ok(()) } else { Err(identity_conflict()) } + } +} + +const IDENTITY_CONFLICT: &str = "bridge_identity_conflict"; + +/// A bridge identity whose command records another tuple: refused, not +/// answered with that command. +fn identity_conflict() -> RuntimeError { + super::refuse(StatusCode::CONFLICT, IDENTITY_CONFLICT) +} + +fn is_identity_conflict(error: &RuntimeError) -> bool { + matches!(error, RuntimeError::Refused { reason, .. } if *reason == IDENTITY_CONFLICT) +} + +/// Whether a write lost to the unique (instance, command type, request id) +/// index: another request recorded the same bridge command first. +fn duplicate_write(error: &RuntimeError) -> bool { + matches!(error, RuntimeError::Database(sqlx::Error::Database(error)) if error.is_unique_violation()) +} + +/// The event that reached a route, after Laravel's request normalization. +struct Event<'a> { + runtime: &'a Runtime, + request: &'a CommandRequest, + body: &'a Value, + adapter: &'a str, + key: &'a str, + target: Value, + correlation: Value, +} + +impl Event<'_> { + fn rejected(&self, action: &str, reason: &str, extra: Value) -> Result<(StatusCode, Value)> { + rejected(self.adapter, Some(action), Some(self.key), reason, extra) + } + + fn context(&self, workflow_id: &str, command: &str, metadata: Value) -> Value { + self.request.context(self.body, workflow_id, command, metadata) + } + + async fn start(&self) -> Result<(StatusCode, Value)> { + let action = "start_workflow"; + let target = &self.target; + if let Some(errors) = errors(target, START_RULES) { + return self.rejected(action, "malformed_payload", json!({"message":"Bridge start target is malformed.", + "errors":errors,"target":redacted(target),"correlation":self.correlation})); + } + let workflow_type = target["workflow_type"].as_str().unwrap_or_default(); + let workflow_id = target["workflow_id"].as_str().map_or_else(|| derived_workflow_id(self.adapter, self.key), str::to_owned); + // `resolveTaskQueue`: a requested queue, or the default one. + let task_queue = target["task_queue"].as_str().map_or("default", str::trim).to_owned(); + let policy = if target["duplicate_policy"] == "reject_duplicate" { "fail" } else { "use-existing" }; + let mut start = json!({"workflow_id":workflow_id,"workflow_type":workflow_type,"task_queue":target["task_queue"], + "business_key":target["business_key"],"input":self.body.get("input"),"duplicate_policy":policy}); + start.as_object_mut().expect("object").retain(|_, value| !value.is_null()); + let context = self.context(&workflow_id, "bridge.start_workflow", + json!({"adapter":self.adapter,"action":action,"idempotency_key":self.key,"task_queue":task_queue})); + // `TaskQueueRoutingGate::workflowStartBlock`, then + // `DefaultWorkflowControlPlane::reserveInstance` namespace first, all + // decided by the store under the transition lock with nothing written + // (`Store::bridge_start`). An instance another namespace holds is + // unavailable, its type and namespace unnamed; PHP checks the type + // first and names it. + let started = match self.runtime.bridge_start(start, context).await { + Ok(started) => started, + Err(RuntimeError::Protocol { response, .. }) if response["reason"] == "validation_failed" => { + return self.rejected(action, "malformed_payload", json!({"message":response["message"], + "target":redacted(target),"correlation":self.correlation})); + } + Err(RuntimeError::Protocol { response, .. }) if response["reason"] == "bridge_start_refused" => { + let decision = &response["bridge"]; + return match decision["refusal"].as_str() { + Some("task_queue_draining") => { + let mut extra = json!({"message":format!("Task queue [{task_queue}] is draining and cannot accept new workflow starts until an active worker cohort is available."), + "target":redacted(&with_workflow_id(target, &workflow_id)),"correlation":self.correlation,"workflow_id":workflow_id, + "workflow_type":workflow_type,"task_queue":task_queue,"rejection_reason":"task_queue_draining", + "control_plane_outcome":"rejected_task_queue_draining"}); + if let (Some(fields), Some(block)) = (extra.as_object_mut(), decision["block"].as_object()) { + fields.extend(block.clone()); + } + self.rejected(action, "task_queue_draining", extra) + } + Some("instance_reserved_for_type") => { + let reserved_type = decision["reserved_type"].as_str().unwrap_or_default(); + self.rejected(action, "unknown_target", json!({"message":format!("Workflow instance [{workflow_id}] is reserved for durable type [{reserved_type}] and cannot be reused for [{workflow_type}]."), + "target":redacted(target),"correlation":self.correlation})) + } + _ => self.rejected(action, "unknown_target", json!({"message":format!("Workflow instance [{workflow_id}] is unavailable."), + "target":redacted(target),"correlation":self.correlation})), + }; + } + Err(error) => return Err(error), + }; + let outcome = started["outcome"].as_str().unwrap_or_default(); + let duplicate = outcome == "returned_existing_active"; + let started_id = started["workflow_id"].as_str().unwrap_or(&workflow_id).to_owned(); + if outcome != "started_new" && !duplicate { + let reason = started["rejection_reason"].as_str().or(started["reason"].as_str()).unwrap_or("unsupported_routing").to_owned(); + return self.rejected(action, &reason, json!({"message":started["message"], + "target":redacted(&with_workflow_id(target, &started_id)),"correlation":self.correlation, + "workflow_id":started_id,"run_id":started["run_id"],"workflow_type":started["workflow_type"], + "rejection_reason":started["rejection_reason"],"control_plane_outcome":outcome})); + } + answer(if duplicate { StatusCode::OK } else { StatusCode::ACCEPTED }, json!({"adapter":self.adapter,"action":action, + "accepted":!duplicate,"outcome":if duplicate { "duplicate" } else { "accepted" }, + "reason":if duplicate { json!("duplicate_start") } else { Value::Null },"idempotency_key":self.key, + "target":redacted(&with_workflow_id(target, &started_id)),"correlation":self.correlation, + "workflow_id":started_id,"run_id":started["run_id"],"workflow_type":started["workflow_type"], + "control_plane_outcome":outcome})) + } + + /// The target of a signal or update, refused when malformed or not a + /// workflow of this namespace. `None` lets the command run; its own + /// transaction answers a duplicate. + async fn command_guard(&self, action: &str, rules: &[(&str, &[Rule])], noun: &str) -> Result> { + let target = &self.target; + if let Some(errors) = errors(target, rules) { + return self.rejected(action, "malformed_payload", json!({"message":format!("Bridge {noun} target is malformed."), + "errors":errors,"target":redacted(target),"correlation":self.correlation})).map(Some); + } + let workflow_id = target["workflow_id"].as_str().unwrap_or_default(); + if !self.runtime.workflow_in_namespace(workflow_id).await? { + return self.rejected(action, "unknown_target", json!({"target":redacted(target),"correlation":self.correlation})).map(Some); + } + Ok(None) + } + + /// The answer for a command this adapter already recorded. + fn duplicate(&self, action: &str, workflow_id: &str, run_id: &Value) -> Result<(StatusCode, Value)> { + answer(StatusCode::OK, json!({"adapter":self.adapter,"action":action,"accepted":false,"outcome":"duplicate", + "idempotency_key":self.key,"target":redacted(&self.target),"correlation":self.correlation, + "workflow_id":workflow_id,"run_id":run_id,"control_plane_outcome":"deduped_existing_command"})) + } + + /// A bridge identity whose stored command records another tuple: a 409 + /// in the bridge envelope, with only the caller's own fields. + fn identity_conflict(&self, action: &str) -> Result<(StatusCode, Value)> { + answer(StatusCode::CONFLICT, json!({"adapter":self.adapter,"action":action,"accepted":false,"outcome":"rejected", + "reason":IDENTITY_CONFLICT,"idempotency_key":self.key, + "message":"The bridge command's identity is recorded for another command.", + "target":redacted(&self.target),"correlation":self.correlation})) + } + + /// A write that lost to the unique index. It is this command's duplicate + /// only when the accepted command under the identity records the same + /// adapter, action, key and target; anything else fails closed. + async fn lost_race(&self, action: &str, kind: &str, key: &BridgeKey<'_>) -> Result<(StatusCode, Value)> { + let workflow_id = self.target["workflow_id"].as_str().unwrap_or_default(); + match self.runtime.bridge_identity_receipt(workflow_id, kind, &key.identity).await? { + Some((run, context)) if key.recorded_in(&context) => self.duplicate(action, workflow_id, &json!(run)), + _ => self.identity_conflict(action), + } + } + + /// The command's bridge key. + fn bridge_key<'k>(&'k self, action: &'k str, field: &'static str) -> BridgeKey<'k> { + BridgeKey::new(self.adapter, action, self.key, field, self.target[field].as_str().unwrap_or_default()) + } + + /// `commandOutcome`: the control plane's answer in the envelope. + fn command_outcome(&self, action: &str, status: StatusCode, result: &Value) -> Result<(StatusCode, Value)> { + if result["reason"] == "instance_not_found" { + return self.rejected(action, "unknown_target", json!({"target":redacted(&self.target),"correlation":self.correlation})); + } + let accepted = status.as_u16() < 400; + answer(if accepted { StatusCode::ACCEPTED } else { StatusCode::UNPROCESSABLE_ENTITY }, json!({ + "adapter":self.adapter,"action":action,"accepted":accepted,"outcome":if accepted { "accepted" } else { "rejected" }, + "reason":if accepted { Value::Null } else { json!(result["reason"].as_str().unwrap_or("unsupported_routing")) }, + "idempotency_key":self.key,"target":redacted(&self.target),"correlation":self.correlation, + "workflow_id":result.get("workflow_id").filter(|id| !id.is_null()).unwrap_or(&self.target["workflow_id"]), + "run_id":result["run_id"],"control_plane_outcome":result["outcome"]})) + } + + /// A refused command, as the control plane's result for it. + fn refused(&self, action: &str, error: RuntimeError, operation: super::control_plane::Operation<'_>, workflow_id: &str) -> Result<(StatusCode, Value)> { + match operation.refusal(error, workflow_id, None) { + RuntimeError::Protocol { status, response } if response["reason"] != "validation_failed" => { + self.command_outcome(action, status, &response) + } + RuntimeError::Refused { status, reason } => self.command_outcome(action, status, &json!({"reason":reason})), + other => Err(other), + } + } + + async fn signal(&self) -> Result<(StatusCode, Value)> { + let action = "signal_workflow"; + if let Some(answer) = self.command_guard(action, SIGNAL_RULES, "signal").await? { + return Ok(answer); + } + let workflow_id = self.target["workflow_id"].as_str().unwrap_or_default(); + let name = self.target["signal_name"].as_str().unwrap_or_default(); + let context = self.context(workflow_id, "bridge.signal_workflow", json!({"adapter":self.adapter,"action":action, + "idempotency_key":self.key,"request_id":self.key,"signal_name":name})); + let body = json!({"input":self.body.get("input")}); + let key = self.bridge_key(action, "signal_name"); + match self.runtime.bridge_signal(workflow_id, name, body, context, &key).await { + Ok(result) if result["outcome"] == "deduped_existing_command" => self.duplicate(action, workflow_id, &result["run_id"]), + Ok(result) => self.command_outcome(action, StatusCode::ACCEPTED, &result), + Err(error) if is_identity_conflict(&error) => self.identity_conflict(action), + Err(error) if duplicate_write(&error) => self.lost_race(action, "signal", &key).await, + Err(error) => self.refused(action, error, super::control_plane::Operation::Signal(name), workflow_id), + } + } + + async fn update(&self) -> Result<(StatusCode, Value)> { + let action = "update_workflow"; + if let Some(answer) = self.command_guard(action, UPDATE_RULES, "update").await? { + return Ok(answer); + } + let workflow_id = self.target["workflow_id"].as_str().unwrap_or_default(); + let name = self.target["update_name"].as_str().unwrap_or_default(); + let wait_for = self.target["wait_for"].as_str().unwrap_or("accepted"); + let context = self.context(workflow_id, "bridge.update_workflow", json!({"adapter":self.adapter,"action":action, + "idempotency_key":self.key,"request_id":self.key,"update_name":name,"wait_for":wait_for})); + // As `WorkflowController::update` admits it. The bridge identity is + // the update's request id; the raw key stays in the metadata. + let mut body = json!({"input":self.body.get("input"),"wait_for":wait_for, + "wait_timeout_seconds":self.target.get("wait_timeout_seconds")}); + body.as_object_mut().expect("object").retain(|_, value| !value.is_null()); + let key = self.bridge_key(action, "update_name"); + match self.runtime.bridge_update(workflow_id, name, body, context, &key).await { + Ok((_, result)) if result["outcome"] == "deduped_existing_command" => self.duplicate(action, workflow_id, &result["run_id"]), + Ok((status, result)) => self.command_outcome(action, status, &result), + Err(error) if is_identity_conflict(&error) => self.identity_conflict(action), + Err(error) if duplicate_write(&error) => self.lost_race(action, "update", &key).await, + Err(error) => self.refused(action, error, super::control_plane::Operation::Update(name), workflow_id), + } + } +} + +/// `POST /api/bridge-adapters/webhook/{adapter}`. +pub(crate) async fn webhook(runtime: &Runtime, request: &CommandRequest, adapter: &str, body: Value) -> Result<(StatusCode, Value)> { + let adapter = adapter.trim(); + if !routing_name(adapter) || adapter.len() > ADAPTER_MAX_BYTES { + return rejected(adapter, None, None, "unsupported_routing", json!({"message":"Bridge adapter route is unsupported."})); + } + let body = super::start_request::normalized(body); + if let Some(errors) = errors(&body, EVENT_RULES) { + return rejected(adapter, None, None, "malformed_payload", json!({"message":"Bridge event payload is malformed.","errors":errors})); + } + let action = body["action"].as_str().unwrap_or_default(); + let key = body["idempotency_key"].as_str().unwrap_or_default(); + let correlation = Some(array_value(&body, "correlation")).filter(|value| !empty(value)).unwrap_or(Value::Null); + let event = Event { runtime, request, body: &body, adapter, key, target: array_value(&body, "target"), correlation }; + match action { + "start_workflow" => event.start().await, + "signal_workflow" => event.signal().await, + "update_workflow" => event.update().await, + _ => event.rejected(action, "unsupported_action", json!({"message":"Bridge action is unsupported.", + "target":redacted(&event.target),"correlation":event.correlation})), + } +} + +store_methods! { + /// `NamespaceWorkflowScope::workflowBound`. + pub(crate) async fn workflow_in_namespace(&self, workflow_id: &str) -> Result { + let found = Self::scalar("SELECT COUNT(*) FROM workflow_instances WHERE id=$1 AND namespace=$2") + .bind(workflow_id) + .bind(self.namespace.as_ref()) + .fetch_one(&self.pool) + .await?; + Ok(found > 0) + } + + /// The accepted command recorded under a bridge identity: its run and + /// its context. + pub(crate) async fn bridge_identity_receipt(&self, workflow_id: &str, kind: &str, identity: &str) -> Result> { + let Some(row) = Self::query("SELECT workflow_run_id,resolved_workflow_run_id,context FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND status='accepted' AND request_id=$3") + .bind(workflow_id).bind(kind).bind(identity).fetch_optional(&self.pool).await? + else { + return Ok(None); + }; + Ok(Some((Self::bridge_run(&row)?, DB::document_row(&row, "context")?))) + } + + /// `duplicateBridgeCommand`, inside the command's transaction and under + /// the transition lock: the run of this bridge command's accepted + /// command, found by its identity, or else by the metadata PHP recorded + /// before takeover. PHP's commands carry no identity: a signal has no + /// request id and an update has the raw key. A command of any status + /// under the identity must record this tuple, or the request fails + /// closed; a rejected one keeps its own receipt (the update path replays + /// it), so only an accepted one is answered here. + pub(super) async fn bridge_receipt(tx: &mut Transaction<'_, DB>, workflow_id: &str, kind: &str, key: &BridgeKey<'_>) -> Result> { + if let Some(row) = Self::query("SELECT status,workflow_run_id,resolved_workflow_run_id,context FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND request_id=$3") + .bind(workflow_id).bind(kind).bind(key.identity.as_str()).fetch_optional(&mut **tx).await? + { + key.verify(DB::optional_document_row(&row, "context")?.as_ref())?; + return if DB::string(&row, "status")? == "accepted" { Ok(Some(Self::bridge_run(&row)?)) } else { Ok(None) }; + } + // PHP orders by `created_at`, which native never writes; an accepted + // command's `accepted_at` is that same instant. + let Some(row) = Self::query(DB::BRIDGE_COMMAND) + .bind(workflow_id).bind(kind).bind(key.adapter).bind(key.action).bind(key.key).bind(key.field).bind(key.name).bind(key.key) + .fetch_optional(&mut **tx).await? + else { + return Ok(None); + }; + Ok(Some(Self::bridge_run(&row)?)) + } + + fn bridge_run(row: &DB::Row) -> Result { + match DB::optional_string(row, "resolved_workflow_run_id")? { + Some(run) => Ok(run), + None => DB::string(row, "workflow_run_id"), + } + } + +} diff --git a/rust/src/runtime/command_requests.rs b/rust/src/runtime/command_requests.rs index 3eabae9e6..dc5984dfa 100644 --- a/rust/src/runtime/command_requests.rs +++ b/rust/src/runtime/command_requests.rs @@ -4,11 +4,18 @@ use super::laravel_validation::{Rule, failure_messages, rule_errors}; use super::{Result, refuse}; use axum::http::StatusCode; use serde_json::Value; -use Rule::{Array, Boolean, In, Integer, Max, Min, Nullable}; +use Rule::{Array, Boolean, In, Integer, Max, Min, Nullable, Pattern}; + +/// Whether a caller's request id stays out of the `dw-bridge:` namespace, +/// which bridge commands' identities own (Server #777). MySQL compares +/// request ids without regard to case, so neither does this. +pub(super) fn unreserved_request_id(request_id: &str) -> bool { + !request_id.trim_start().get(..10).is_some_and(|prefix| prefix.eq_ignore_ascii_case("dw-bridge:")) +} const SIGNAL_RULES: &[(&str, &[Rule])] = &[ ("input", &[Nullable, Array]), - ("request_id", &[Nullable, Rule::String, Max(255)]), + ("request_id", &[Nullable, Rule::String, Max(255), Pattern(unreserved_request_id)]), ("dry_run", &[Nullable, Boolean]), ]; @@ -19,7 +26,7 @@ pub(super) const UPDATE_WAIT_TIMEOUT_SECONDS_MAX: i64 = crate::config_audit::UPD const UPDATE_RULES: &[(&str, &[Rule])] = &[ ("input", &[Nullable, Array]), - ("request_id", &[Nullable, Rule::String, Max(255)]), + ("request_id", &[Nullable, Rule::String, Max(255), Pattern(unreserved_request_id)]), ("wait_for", &[Nullable, Rule::String, In(&["accepted", "completed"])]), ("wait_timeout_seconds", &[Nullable, Integer, Min(1), Max(UPDATE_WAIT_TIMEOUT_SECONDS_MAX)]), ]; @@ -90,6 +97,13 @@ mod tests { errors(update(&json!({"wait_for":"completed","wait_timeout_seconds":UPDATE_WAIT_TIMEOUT_SECONDS_MAX + 1}))), json!({"wait_timeout_seconds":["The wait timeout seconds field must not be greater than 60."]}) ); + for reserved in ["dw-bridge:v1:00", "DW-Bridge:anything", " dw-bridge:"] { + let refused = json!({"request_id":["The request id field format is invalid."]}); + assert_eq!(errors(update(&json!({"request_id":reserved}))), refused, "{reserved}"); + assert_eq!(errors(signal(&json!({"request_id":reserved}))), refused, "{reserved}"); + } + assert!(update(&json!({"request_id":"dw-bridge"})).is_ok()); + assert!(update(&json!({"request_id":"my-dw-bridge:key"})).is_ok()); assert_eq!( errors(update(&json!({"wait_policy":"accepted"}))), json!({"wait_policy":["The wait_policy field is no longer supported. Use wait_for."]}) diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index a504edd29..9b9887e05 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -508,6 +508,14 @@ impl Runtime { } pub(crate) async fn start(&self, body: Value, context: Option) -> Result { + self.start_command(body, context, false).await + } + /// A bridge adapter's start: the store decides the draining queue and the + /// instance reservation under the transition lock (`Store::bridge_start`). + pub(crate) async fn bridge_start(&self, body: Value, context: Value) -> Result { + self.start_command(body, Some(context), true).await + } + async fn start_command(&self, body: Value, context: Option, bridge: bool) -> Result { // PHP validates the request before it reads the memo or the input. let mut body = super::start_request::normalized(body); super::input_envelope::transport(&body)?; @@ -526,10 +534,13 @@ impl Runtime { let memo = super::memos::prepare_initial(self.signal_codec.clone(), body["memo"].clone()).await?; let external = self.prepare_external(&body["input"]).await?; - match &*self.storage { - Storage::Sqlite(store) => store.start(body, memo, external, context).await, - Storage::Postgres(store) => store.start(body, memo, external, context).await, - Storage::MySql(store) => store.start(body, memo, external, context).await, + match (&*self.storage, bridge) { + (Storage::Sqlite(store), false) => store.start(body, memo, external, context).await, + (Storage::Postgres(store), false) => store.start(body, memo, external, context).await, + (Storage::MySql(store), false) => store.start(body, memo, external, context).await, + (Storage::Sqlite(store), true) => store.bridge_start(body, memo, external, context).await, + (Storage::Postgres(store), true) => store.bridge_start(body, memo, external, context).await, + (Storage::MySql(store), true) => store.bridge_start(body, memo, external, context).await, } } /// `ActivityController::start`: validates like PHP, then starts the @@ -792,6 +803,24 @@ impl Runtime { name: &str, body: Value, context: Option, + ) -> Result { + self.signal_command(workflow_id, run_id, name, body, context, None).await + } + /// A bridge adapter's signal, recorded under its bridge identity. A + /// command already recorded under it is answered as + /// `deduped_existing_command`. + pub(crate) async fn bridge_signal(&self, workflow_id: &str, name: &str, body: Value, context: Value, + key: &super::bridge_adapters::BridgeKey<'_>) -> Result { + self.signal_command(workflow_id, None, name, body, Some(context), Some(key)).await + } + async fn signal_command( + &self, + workflow_id: &str, + run_id: Option<&str>, + name: &str, + body: Value, + context: Option, + bridge: Option<&super::bridge_adapters::BridgeKey<'_>>, ) -> Result { let body = super::start_request::normalized(body); super::input_envelope::transport(&body)?; @@ -828,6 +857,7 @@ impl Runtime { context, self.structural, external.as_ref(), + bridge, ) .await } @@ -843,6 +873,7 @@ impl Runtime { context, self.structural, external.as_ref(), + bridge, ) .await } @@ -858,6 +889,7 @@ impl Runtime { context, self.structural, external.as_ref(), + bridge, ) .await } @@ -924,6 +956,24 @@ impl Runtime { name: &str, body: Value, context: Option, + ) -> Result<(StatusCode, Value)> { + self.update_command(workflow_id, run_id, name, body, context, None).await + } + /// A bridge adapter's update, whose request id is its bridge identity. + /// An accepted command already recorded under it is answered as + /// `deduped_existing_command`. + pub(crate) async fn bridge_update(&self, workflow_id: &str, name: &str, body: Value, context: Value, + key: &super::bridge_adapters::BridgeKey<'_>) -> Result<(StatusCode, Value)> { + self.update_command(workflow_id, None, name, body, Some(context), Some(key)).await + } + async fn update_command( + &self, + workflow_id: &str, + run_id: Option<&str>, + name: &str, + body: Value, + context: Option, + bridge: Option<&super::bridge_adapters::BridgeKey<'_>>, ) -> Result<(StatusCode, Value)> { let body = super::start_request::normalized(body); super::input_envelope::transport(&body)?; @@ -949,17 +999,17 @@ impl Runtime { match &*self.storage { Storage::Sqlite(store) => { store - .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref()) + .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref(), bridge) .await } Storage::Postgres(store) => { store - .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref()) + .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref(), bridge) .await } Storage::MySql(store) => { store - .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref()) + .update_workflow(workflow_id, run_id, name, body, prepared, context, self.signal_codec.clone(), self.structural, external.as_ref(), bridge) .await } } @@ -1607,6 +1657,8 @@ impl Runtime { delegate!(heartbeat_activity(task_id: &str, body: Value, protocol: &str) -> Result); delegate!(activity_status(task_id: &str, body: Value, protocol: &str) -> Result); delegate!(renew_activity_lease(task_id: &str, body: Value) -> Result); + delegate!(workflow_in_namespace(workflow_id: &str) -> Result); + delegate!(bridge_identity_receipt(workflow_id: &str, kind: &str, identity: &str) -> Result>); delegate!(nexus_operations(workflow_id: &str, run_id: Option<&str>, limit: i64) -> Result>); delegate!(import_existing_run(run_id: &str) -> Result>); delegate!(import_write(plan: &super::workflow_import::Plan) -> std::result::Result); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index fd0571f1e..667cc8622 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -67,6 +67,7 @@ pub fn router(runtime: Runtime) -> Router { delete(delete_search_attribute), ) .route("/api/workflows", get(list_workflows).post(start)) + .route("/api/bridge-adapters/webhook/{adapter}", post(bridge_webhook)) .route("/api/workflows/import/embedded-v2", post(import_embedded_v2)) .route("/api/workflows/import/waterline-v1", post(import_waterline_v1)) .route("/api/activities", get(list_activities).post(start_activity)) @@ -669,7 +670,7 @@ enum Rest { } /// Route families a named namespace may use, by prefix. -const NAMESPACE_FAMILIES: [(&str, Rest, Allowed); 9] = [ +const NAMESPACE_FAMILIES: [(&str, Rest, Allowed); 10] = [ // Runtime credentials are server-wide; the namespace only scopes a tenant. ("/api/runtime-credentials/", Rest::Anything, Allowed::Any), ("/api/external-payloads/v1/", Rest::Anything, READ), @@ -685,6 +686,8 @@ const NAMESPACE_FAMILIES: [(&str, Rest, Allowed); 9] = [ ("/api/workers/", Rest::OneSegment, Allowed::Only(&["GET", "DELETE"])), // Deployments are the namespace's own build-id rollouts. ("/api/deployments/", Rest::NonEmpty, READ_WRITE), + // A bridge event acts on workflows of the requested namespace. + ("/api/bridge-adapters/webhook/", Rest::OneSegment, WRITE), // Worker session visibility reads the namespace's own sessions. ("/api/worker-sessions/", Rest::OneSegment, READ), ]; @@ -1116,6 +1119,19 @@ async fn start( Ok((status, Json(super::control_plane::Operation::Start.response(result)))) } +/// `POST /api/bridge-adapters/webhook/{adapter}`. PHP validates the body +/// itself, so a body that is not a JSON object reads as an empty one. +async fn bridge_webhook( + Extension(runtime): Extension, + command: CommandRequest, + Path(adapter): Path, + body: axum::body::Bytes, +) -> Result<(StatusCode, Json)> { + let body = serde_json::from_slice::(&body).ok().filter(Value::is_object).unwrap_or_else(|| json!({})); + let (status, answer) = super::bridge_adapters::webhook(&runtime, &command, &adapter, body).await?; + Ok((status, Json(answer))) +} + /// `POST /api/workflows/import/embedded-v2`. The controller reads the body /// and the query string itself, as Laravel's request input. async fn import_embedded_v2( diff --git a/rust/src/runtime/message_streams.rs b/rust/src/runtime/message_streams.rs index 2f3294731..f50b7ac1a 100644 --- a/rust/src/runtime/message_streams.rs +++ b/rust/src/runtime/message_streams.rs @@ -650,7 +650,8 @@ where let message_sequence = DB::number(&instance, "last_message_sequence")? + 1; let workflow_type = DB::string(&run, "workflow_type")?; let write = SignalWrite { workflow_id, run_id: &run_id, workflow_type: &workflow_type, name: INTERNAL_SIGNAL, - command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context }; + command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context, + request_id: None }; let pending = Self::scalar("SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$1 AND status='received'") .bind(&run_id).fetch_one(&mut **tx).await?; let exceeded = super::memo_limits::Exceeded::pending_signals(limits.pending_signals, pending) diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index dc0595f9d..9f05a7dcf 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -36,6 +36,7 @@ mod activity_results; mod activity_summary; mod all_groups; mod backend; +mod bridge_adapters; mod bundle_verifier; mod cancellation; mod cancellation_cascade; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 8af3a7def..4bebaffac 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -86,6 +86,8 @@ pub(super) struct SignalWrite<'a> { pub(super) command_sequence: i64, pub(super) message_sequence: i64, pub(super) context: &'a Value, + /// A bridge command's identity, as its `request_id`. + pub(super) request_id: Option<&'a str>, } /// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`). const MESSAGE_STREAM_SIGNAL: &str = super::message_streams::INTERNAL_SIGNAL; @@ -331,6 +333,33 @@ where memo: super::memos::PreparedMemo, external: Option, context: Option, + ) -> Result { + self.start_command(body, memo, external, context, false).await + } + + /// A bridge adapter's start (Server #776). Under the transition lock and + /// before anything is written, in the bridge's order: a draining queue, + /// then an instance another namespace holds, then one reserved for + /// another type. Each is refused as a `Protocol` whose `bridge` field + /// names the decision, so no state change between a read and the write + /// can turn into another answer. + pub(crate) async fn bridge_start( + &self, + body: Value, + memo: super::memos::PreparedMemo, + external: Option, + context: Option, + ) -> Result { + self.start_command(body, memo, external, context, true).await + } + + async fn start_command( + &self, + body: Value, + memo: super::memos::PreparedMemo, + external: Option, + context: Option, + bridge: bool, ) -> Result { let mut tx = self.begin().await?; let arguments = external.as_ref().map(|prepared| prepared.stored()); @@ -338,18 +367,37 @@ where self.claim_external_in(&mut tx, &prepared.reference, &prepared.uri) .await?; } + let request = super::start_request::validate(&body)?; + if bridge { + let refusal = |decision: Value| super::RuntimeError::Protocol { status: StatusCode::CONFLICT, + response: json!({"reason":"bridge_start_refused","bridge":decision}) }; + if let Some(block) = Self::routing_block(&mut tx, &self.namespace, &request.task_queue, self.worker_stale_after_seconds).await? { + return Err(refusal(json!({"refusal":"task_queue_draining","block":block}))); + } + if let Some(workflow_id) = &request.workflow_id + && let Some(instance) = Self::query("SELECT workflow_type,namespace FROM workflow_instances WHERE id=$1") + .bind(workflow_id.as_str()).fetch_optional(&mut *tx).await? + { + if DB::optional_string(&instance, "namespace")?.as_deref() != Some(self.namespace.as_ref()) { + return Err(refusal(json!({"refusal":"instance_unavailable"}))); + } + let reserved_type = DB::string(&instance, "workflow_type")?; + if reserved_type != request.workflow_type { + return Err(refusal(json!({"refusal":"instance_reserved_for_type","reserved_type":reserved_type}))); + } + } + } // `TaskQueueRoutingGate`: a queue whose live workers all drain takes // no new start. A workflow id reserved in another namespace is // refused for that first. - let request = super::start_request::validate(&body)?; let reserved = match &request.workflow_id { - Some(workflow_id) => Self::query("SELECT namespace FROM workflow_instances WHERE id=$1") + Some(workflow_id) if !bridge => Self::query("SELECT namespace FROM workflow_instances WHERE id=$1") .bind(workflow_id.as_str()).fetch_optional(&mut *tx).await? .map(|instance| DB::optional_string(&instance, "namespace")).transpose()?.flatten() .is_some_and(|reserved| reserved != self.namespace.as_ref()), - None => false, + _ => false, }; - if !reserved && let Some(block) = Self::routing_block(&mut tx, &self.namespace, &request.task_queue, self.worker_stale_after_seconds).await? { + if !bridge && !reserved && let Some(block) = Self::routing_block(&mut tx, &self.namespace, &request.task_queue, self.worker_stale_after_seconds).await? { let mut response = json!({"workflow_id":request.workflow_id,"command_status":"rejected","command_source":"control_plane", "outcome":"rejected_task_queue_draining","reason":"task_queue_draining","rejection_reason":"task_queue_draining", "message":format!("Task queue [{}] is draining and cannot accept new workflow starts until an active worker cohort is available.", request.task_queue), @@ -808,6 +856,7 @@ where context: Option, limits: super::memo_limits::StructuralLimits, external: Option<&super::external_resolution::PreparedExternal>, + bridge: Option<&super::bridge_adapters::BridgeKey<'_>>, ) -> Result { // The authenticated request's context; a direct call records only // the request id and signal name. @@ -837,6 +886,13 @@ where let instance = Self::query("SELECT current_run_id,last_message_sequence FROM workflow_instances WHERE id=$1 AND namespace=$2") .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?.ok_or_else(|| refuse(StatusCode::NOT_FOUND, "instance_not_found"))?; let current_run = DB::string(&instance, "current_run_id")?; + // A bridge signal already accepted is answered, under the transition + // lock, before anything else is admitted. + if let Some(key) = bridge + && let Some(run) = Self::bridge_receipt(&mut tx, workflow_id, "signal", key).await? + { + return Ok(json!({"workflow_id":workflow_id,"run_id":run,"outcome":"deduped_existing_command"})); + } if run_id.is_some_and(|selected| selected != current_run) { return Err(refuse( StatusCode::CONFLICT, @@ -979,7 +1035,8 @@ where ) .await?; let write = SignalWrite { workflow_id, run_id: ¤t_run, workflow_type: &workflow_type, name, - command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context: &context }; + command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context: &context, + request_id: None }; Self::record_rejected_signal(&mut tx, &write, outcome, reason, &prepared.command, &row_arguments, &row_errors).await?; tx.commit().await?; let mut response = json!({"workflow_id":workflow_id,"run_id":current_run,"resolved_run_id":current_run, @@ -997,7 +1054,8 @@ where } let workflow_type = DB::string(&run, "workflow_type")?; let write = SignalWrite { workflow_id, run_id: ¤t_run, workflow_type: &workflow_type, name, - command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context: &context }; + command_id: &command_id, signal_id: &signal_id, command_sequence, message_sequence, context: &context, + request_id: bridge.map(|key| key.identity.as_str()) }; Self::record_accepted_signal(&mut tx, &write, &prepared.command, &prepared.arguments, &DB::string(&run, "queue")?).await?; tx.commit().await?; self.wake.notify_waiters(); @@ -1051,10 +1109,10 @@ where } } let accepted_at = now(); - Self::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,resolved_workflow_run_id,command_type,source,status,outcome,workflow_type,payload_codec,payload,command_sequence,message_sequence,accepted_at,context) VALUES ($1,$2,$3,$4,'signal','control_plane','accepted','signal_received',$5,'avro',$6,$7,$8,$9,$10)") + Self::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,resolved_workflow_run_id,command_type,source,status,outcome,workflow_type,payload_codec,payload,command_sequence,message_sequence,accepted_at,context,request_id) VALUES ($1,$2,$3,$4,'signal','control_plane','accepted','signal_received',$5,'avro',$6,$7,$8,$9,$10,$11)") .bind(write.command_id).bind(write.workflow_id).bind(write.run_id).bind(write.run_id).bind(write.workflow_type) .bind(command).bind(write.command_sequence).bind(write.message_sequence).bind(DB::bind_time(accepted_at)) - .bind(DB::document(write.context)).execute(&mut **tx).await?; + .bind(DB::document(write.context)).bind(write.request_id).execute(&mut **tx).await?; Self::query("INSERT INTO workflow_signal_records(id,workflow_command_id,workflow_instance_id,workflow_run_id,resolved_workflow_run_id,signal_name,signal_wait_id,status,outcome,command_sequence,payload_codec,arguments,received_at) VALUES ($1,$2,$3,$4,$5,$6,$7,'received','signal_received',$8,'avro',$9,$10)") .bind(write.signal_id).bind(write.command_id).bind(write.workflow_id).bind(write.run_id).bind(write.run_id).bind(write.name).bind(&wait_id) .bind(write.command_sequence).bind(arguments).bind(DB::bind_time(accepted_at)).execute(&mut **tx).await?; diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 9100a1d7a..edec4ac40 100644 --- a/rust/src/runtime/updates.rs +++ b/rust/src/runtime/updates.rs @@ -345,8 +345,9 @@ where codec: Arc, limits: super::memo_limits::StructuralLimits, external: Option<&super::external_resolution::PreparedExternal>, + bridge: Option<&super::bridge_adapters::BridgeKey<'_>>, ) -> Result<(StatusCode, Value)> { - self.admit_update(workflow_id, selected_run, name, body, prepared, context, codec, limits, external, None) + self.admit_update(workflow_id, selected_run, name, body, prepared, context, codec, limits, external, None, bridge) .await } @@ -365,6 +366,7 @@ where limits: super::memo_limits::StructuralLimits, external: Option<&super::external_resolution::PreparedExternal>, approval: Option, + bridge: Option<&super::bridge_adapters::BridgeKey<'_>>, ) -> Result<(StatusCode, Value)> { // PHP's update_name column holds 255 bytes, one fewer than its // validation allows; native refuses what it cannot store. @@ -377,7 +379,8 @@ where // `WorkflowController::update`'s rules, on the trimmed request. let request = super::start_request::normalized(body.clone()); let rules: &[(&str, &[Rule])] = &[ - ("request_id", &[Rule::Nullable, Rule::String, Rule::Max(255)]), + ("request_id", &[Rule::Nullable, Rule::String, Rule::Max(255), + Rule::Pattern(super::command_requests::unreserved_request_id)]), ("wait_for", &[Rule::Nullable, Rule::String, Rule::In(&["accepted", "completed"])]), ("wait_timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(1), Rule::Max(super::command_requests::UPDATE_WAIT_TIMEOUT_SECONDS_MAX)]), @@ -385,7 +388,9 @@ where if let Some(error) = failure_messages(rule_errors(&request, rules, |_, _| None)) { return Err(error); } - let request_id = request["request_id"].as_str(); + // A bridge update's request id is its bridge identity, which no + // caller may send. + let request_id = bridge.map(|key| key.identity.as_str()).or_else(|| request["request_id"].as_str()); // PHP waits for acceptance unless the request asks for completion. let wait_for = request["wait_for"].as_str().unwrap_or("accepted"); let timeout = super::laravel_validation::integer(&request["wait_timeout_seconds"]); @@ -416,11 +421,26 @@ where return Ok((status, validation_response(failure, wait_for, context.as_ref(), name))); } let existing = if let Some(request_id) = request_id { - Self::query("SELECT u.id,u.update_name,u.workflow_run_id FROM workflow_updates u JOIN workflow_commands c ON c.id=u.workflow_command_id AND c.workflow_instance_id=u.workflow_instance_id WHERE c.workflow_instance_id=$1 AND c.command_type='update' AND c.request_id=$2") + Self::query("SELECT u.id,u.update_name,u.workflow_run_id,c.context FROM workflow_updates u JOIN workflow_commands c ON c.id=u.workflow_command_id AND c.workflow_instance_id=u.workflow_instance_id WHERE c.workflow_instance_id=$1 AND c.command_type='update' AND c.request_id=$2") .bind(workflow_id).bind(request_id).fetch_optional(&mut *tx).await? } else { None }; + // An update under a bridge identity, accepted or rejected, is this + // bridge command's only when it records the same adapter, action, + // raw key and target; anything else fails closed before its receipt + // is answered or replayed. + if let (Some(key), Some(existing)) = (bridge, &existing) { + key.verify(DB::optional_document_row(existing, "context")?.as_ref())?; + } + // A bridge update already accepted is answered, under the transition + // lock, as the bridge's duplicate. A rejected one keeps the ordinary + // receipt below, as PHP replays it. + if let Some(key) = bridge + && let Some(run) = Self::bridge_receipt(&mut tx, workflow_id, "update", key).await? + { + return Ok((StatusCode::OK, json!({"workflow_id":workflow_id,"run_id":run,"outcome":"deduped_existing_command"}))); + } let update_id = if let Some(existing) = existing { // Look up the original receipt before validating changed arguments // or admitting another mutation. No result or event is rewritten. @@ -525,7 +545,7 @@ where } let approval = Approval { task_id, run_id: run_id.clone(), input_hash }; Box::pin(self.admit_update(workflow_id, selected_run, name, body, prepared, context, codec, - limits, external, Some(approval))) + limits, external, Some(approval), bridge)) .await } super::update_validation::Validation::Rejected { failure, validation_errors } => { diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 816707a7d..cf9f88e60 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -204,6 +204,8 @@ mod activity_timeout_pass; mod start_request; #[path = "execution/command_requests.rs"] mod command_requests; +#[path = "execution/bridge_adapters.rs"] +mod bridge_adapters; #[path = "execution/describe_actions.rs"] mod describe_actions; #[path = "execution/run_deadlines.rs"] diff --git a/rust/tests/execution/bridge_adapters.rs b/rust/tests/execution/bridge_adapters.rs new file mode 100644 index 000000000..79703de82 --- /dev/null +++ b/rust/tests/execution/bridge_adapters.rs @@ -0,0 +1,450 @@ +use super::*; + +// Bridge adapter webhooks (Server #776, #777): one effect and one receipt per +// bridge command, even under contention, and a command identity that keeps +// apart the same raw key from another adapter or for another target. + +async fn bridge(app: &Router, adapter: &str, body: Value) -> (StatusCode, Value) { + request(app, "POST", &format!("/api/bridge-adapters/webhook/{adapter}"), body).await +} + +fn events(history: &Value, event_type: &str) -> usize { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == event_type).count() +} + +fn signal_event(key: &str, workflow_id: &str, name: &str) -> Value { + json!({"action":"signal_workflow","idempotency_key":key,"target":{"workflow_id":workflow_id,"signal_name":name},"input":[1]}) +} + +fn update_event(key: &str, workflow_id: &str, name: &str) -> Value { + json!({"action":"update_workflow","idempotency_key":key,"target":{"workflow_id":workflow_id,"update_name":name},"input":[1]}) +} + +/// A worker whose workflow declares the updates `set` and `add`. +async fn two_update_worker(app: &Router) { + let parameters = json!([{"name":"value","type":"int","required":true,"variadic":false,"default_available":false,"allows_null":false}]); + let (status, body) = request(app, "POST", "/api/worker/register", json!({ + "worker_id":"bridge-updates","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":["workflow_updates"],"workflow_command_contracts":{"echo":{ + "signals":["finish"],"signal_contracts":[{"name":"finish","parameters":parameters}], + "updates":["set","add"],"update_validators":[], + "update_contracts":[{"name":"set","parameters":parameters},{"name":"add","parameters":parameters}] + }}})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +/// The same event sent eight times at once, through two pools. +async fn contend(apps: &[Router; 2], adapter: &'static str, event: Value) -> Vec<(StatusCode, Value)> { + let mut tasks = tokio::task::JoinSet::new(); + for index in 0..8 { + let app = apps[index % 2].clone(); + let event = event.clone(); + tasks.spawn(async move { bridge(&app, adapter, event).await }); + } + let mut answers = Vec::new(); + while let Some(answer) = tasks.join_next().await { + answers.push(answer.unwrap()); + } + answers +} + +/// One acceptance and seven duplicates, all naming the same run. +fn one_receipt(answers: &[(StatusCode, Value)], run_id: &Value) { + let accepted = answers.iter().filter(|(status, answer)| *status == StatusCode::ACCEPTED && answer["outcome"] == "accepted").count(); + let duplicates: Vec<_> = answers.iter().filter(|(status, answer)| *status == StatusCode::OK && answer["outcome"] == "duplicate").collect(); + assert_eq!((accepted, duplicates.len()), (1, 7), "{answers:?}"); + for (_, duplicate) in duplicates { + assert_eq!(duplicate["control_plane_outcome"], "deduped_existing_command", "{duplicate}"); + assert_eq!(&duplicate["run_id"], run_id, "{duplicate}"); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_identical_bridge_signals_apply_once() { + let database = TestDatabase::new().await; + let apps = [router(database.open().await.unwrap()), router(database.open().await.unwrap())]; + register(&apps[0], "worker", json!(["echo"]), json!([])).await; + let started = start(&apps[0], "bridge-signal-race").await; + let answers = contend(&apps, "acme", signal_event("evt-1", "bridge-signal-race", "note")).await; + one_receipt(&answers, &started["run_id"]); + let history = run_history(&apps[0], &started["workflow_id"], &started["run_id"]).await; + assert_eq!(events(&history, "SignalReceived"), 1, "{history}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_identical_bridge_updates_apply_once() { + let database = TestDatabase::new().await; + let apps = [router(database.open().await.unwrap()), router(database.open().await.unwrap())]; + two_update_worker(&apps[0]).await; + let started = start(&apps[0], "bridge-update-race").await; + open_update_wait(&apps[0], "bridge-updates").await; + let answers = contend(&apps, "acme", update_event("evt-1", "bridge-update-race", "set")).await; + one_receipt(&answers, &started["run_id"]); + let history = run_history(&apps[0], &started["workflow_id"], &started["run_id"]).await; + assert_eq!(events(&history, "UpdateAccepted"), 1, "{history}"); +} + +// The same raw key is another command from another adapter or for another +// target name: each has its own effect and receipt. +#[tokio::test] +async fn a_raw_key_is_scoped_by_adapter_and_target_name() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + let signalled = start(&app, "bridge-signal-scope").await; + for (adapter, name) in [("acme", "note"), ("other", "note"), ("acme", "memo")] { + let (status, answer) = bridge(&app, adapter, signal_event("shared-key", "bridge-signal-scope", name)).await; + assert_eq!((status, &answer["outcome"]), (StatusCode::ACCEPTED, &json!("accepted")), "{adapter} {name}: {answer}"); + } + let (status, again) = bridge(&app, "other", signal_event("shared-key", "bridge-signal-scope", "note")).await; + assert_eq!((status, &again["outcome"]), (StatusCode::OK, &json!("duplicate")), "{again}"); + let history = run_history(&app, &signalled["workflow_id"], &signalled["run_id"]).await; + assert_eq!(events(&history, "SignalReceived"), 3, "{history}"); + + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + two_update_worker(&app).await; + let updated = start(&app, "bridge-update-scope").await; + open_update_wait(&app, "bridge-updates").await; + // With the raw key as the request id, the second adapter's update would + // reuse the first one's receipt, and the second name would be refused as + // a retargeted request. + for (adapter, name) in [("acme", "set"), ("other", "set"), ("acme", "add")] { + let (status, answer) = bridge(&app, adapter, update_event("shared-key", "bridge-update-scope", name)).await; + assert_eq!((status, &answer["outcome"]), (StatusCode::ACCEPTED, &json!("accepted")), "{adapter} {name}: {answer}"); + } + let (status, again) = bridge(&app, "acme", update_event("shared-key", "bridge-update-scope", "add")).await; + assert_eq!((status, &again["outcome"]), (StatusCode::OK, &json!("duplicate")), "{again}"); + let history = run_history(&app, &updated["workflow_id"], &updated["run_id"]).await; + assert_eq!(events(&history, "UpdateAccepted"), 3, "{history}"); + // The raw key stays in the command metadata; the request id is the + // fixed-length bridge identity. + let accepted: Vec<_> = history.as_array().unwrap().iter().filter(|event| event["event_type"] == "UpdateAccepted").collect(); + for event in accepted { + let request_id = event["payload"]["request_id"].as_str().unwrap_or_default(); + assert!(request_id.starts_with("dw-bridge:v1:") && request_id.len() == 77, "{event}"); + } +} + +// A derived workflow id is `bridge--<32 hex>`, so the adapter is at +// most 64 bytes and the id fits 128. A longer adapter persists nothing. +#[tokio::test] +async fn the_adapter_name_is_bounded_before_anything_is_persisted() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let start_event = json!({"action":"start_workflow","idempotency_key":"evt-long","target":{"workflow_type":"echo","task_queue":"test"}}); + let longest = "a".repeat(64); + let (status, accepted) = bridge(&app, &longest, start_event.clone()).await; + assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); + assert!(accepted["workflow_id"].as_str().unwrap().len() <= 128, "{accepted}"); + let too_long = "a".repeat(65); + let (status, refused) = bridge(&app, &too_long, start_event).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{refused}"); + assert_eq!(refused["reason"], "unsupported_routing", "{refused}"); + let (status, list) = request(&app, "GET", "/api/workflows", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{list}"); + let ids: Vec<_> = list["workflows"].as_array().unwrap().iter().map(|workflow| workflow["workflow_id"].clone()).collect(); + assert_eq!(ids, vec![accepted["workflow_id"].clone()], "{list}"); +} + +// A caller cannot send a request id in the bridge identities' namespace, so +// no regular signal or update can claim a bridge command's identity first. +#[tokio::test] +async fn regular_routes_refuse_the_bridge_request_id_namespace() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + two_update_worker(&app).await; + let started = start(&app, "bridge-preclaim").await; + open_update_wait(&app, "bridge-updates").await; + let run = started["run_id"].as_str().unwrap(); + for (path, body) in [ + ("/api/workflows/bridge-preclaim/update/set".to_owned(), json!({"input":[1],"request_id":"dw-bridge:v1:0000"})), + (format!("/api/workflows/bridge-preclaim/runs/{run}/update/set"), json!({"input":[1],"request_id":"DW-BRIDGE:v1:0000"})), + ("/api/workflows/bridge-preclaim/signal/finish".to_owned(), json!({"input":[1],"request_id":"dw-bridge:v1:0000"})), + (format!("/api/workflows/bridge-preclaim/runs/{run}/signal/finish"), json!({"input":[1],"request_id":"dw-bridge:"})), + ] { + let (status, refused) = request(&app, "POST", &path, body).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{path}: {refused}"); + assert_eq!(refused["errors"]["request_id"], json!(["The request id field format is invalid."]), "{path}: {refused}"); + } + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!((events(&history, "UpdateAccepted"), events(&history, "SignalReceived")), (0, 0), "{history}"); + // The bridge's own update is admitted under its identity. + let (status, answer) = bridge(&app, "acme", update_event("evt-1", "bridge-preclaim", "set")).await; + assert_eq!(status, StatusCode::ACCEPTED, "{answer}"); +} + +// Security holds on 867a2d53 (#776): a command already stored under a bridge +// identity, accepted or rejected, answers the bridge only when it records the +// same adapter, action, raw key and target. Before the prefix reservation a +// caller could choose the predictable identity, so these rows are seeded by +// rewriting a regular command's request id. + +fn identity(adapter: &str, action: &str, key: &str, field: &str, name: &str) -> String { + use sha2::{Digest, Sha256}; + let mut digest = Sha256::new(); + for part in [adapter, action, key, field, name] { + digest.update((part.len() as u64).to_be_bytes()); + digest.update(part.as_bytes()); + } + let hex: String = digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect(); + format!("dw-bridge:v1:{hex}") +} + +/// Gives the regular command `filter` selects a bridge identity as its +/// request id, optionally recording it as rejected. +async fn preclaim(database: &TestDatabase, filter: &str, identity: &str, rejected: bool) { + let status = if rejected { ",status='rejected'" } else { "" }; + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_commands SET request_id='{identity}'{status} WHERE {filter}"))).await; +} + +async fn count(database: &TestDatabase, sql: String) -> i64 { + task_retry::rows(database, Box::leak(sql.into_boxed_str())).await +} + +/// The bridge's 409 for an identity recorded for another command: its own +/// envelope with the caller's fields only, nothing stored and no error text. +fn identity_conflict(answer: &Value, action: &str, key: &str, target: Value) { + assert_eq!(answer, &json!({"schema":"durable-workflow.v2.bridge-adapter-outcome.contract","version":1, + "adapter":"acme","action":action,"accepted":false,"outcome":"rejected","reason":"bridge_identity_conflict", + "idempotency_key":key,"message":"The bridge command's identity is recorded for another command.","target":target})); +} + +#[tokio::test] +async fn a_rejected_update_under_the_identity_is_not_replayed_for_another_tuple() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + two_update_worker(&app).await; + let started = start(&app, "bridge-rejected-claim").await; + let (status, terminated) = request(&app, "POST", "/api/workflows/bridge-rejected-claim/terminate", json!({})).await; + assert!(status.is_success(), "{terminated}"); + // A regular update to the closed run is rejected durably under its + // request id, which then takes the bridge command's identity. + let (status, rejected) = request(&app, "POST", "/api/workflows/bridge-rejected-claim/update/set", + json!({"input":[1],"request_id":"seed-rejected"})).await; + assert!(status.is_client_error(), "{rejected}"); + let claimed = identity("acme", "update_workflow", "evt-r", "update_name", "set"); + preclaim(&database, "request_id='seed-rejected'", &claimed, false).await; + let updates = "SELECT COUNT(*) FROM workflow_updates WHERE workflow_instance_id='bridge-rejected-claim'".to_owned(); + let commands = "SELECT COUNT(*) FROM workflow_commands WHERE workflow_instance_id='bridge-rejected-claim'".to_owned(); + let before = (count(&database, updates.clone()).await, count(&database, commands.clone()).await); + assert_eq!(count(&database, format!("SELECT COUNT(*) FROM workflow_commands WHERE request_id='{claimed}' AND status='rejected'")).await, 1); + + let (status, answer) = bridge(&app, "acme", update_event("evt-r", "bridge-rejected-claim", "set")).await; + assert_eq!(status, StatusCode::CONFLICT, "{answer}"); + identity_conflict(&answer, "update_workflow", "evt-r", json!({"workflow_id":"bridge-rejected-claim","update_name":"set"})); + assert_eq!((count(&database, updates).await, count(&database, commands).await), before); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!(events(&history, "UpdateAccepted"), 0, "{history}"); +} + +#[tokio::test] +async fn a_signal_under_the_identity_answers_only_its_own_tuple() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + let started = start(&app, "bridge-signal-claim").await; + // An accepted regular signal, and one recorded as rejected, each under + // a bridge command's identity. Native keeps a signal's request id in its + // context, so the stored command, as PHP could have left it, is + // selected as the instance's one signal without a request id. + for (key, rejected) in [("evt-a", false), ("evt-b", true)] { + let (status, body) = request(&app, "POST", "/api/workflows/bridge-signal-claim/signal/note", json!({"input":[1]})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + preclaim(&database, "workflow_instance_id='bridge-signal-claim' AND command_type='signal' AND request_id IS NULL", + &identity("acme", "signal_workflow", key, "signal_name", "note"), rejected).await; + } + let commands = "SELECT COUNT(*) FROM workflow_commands WHERE workflow_instance_id='bridge-signal-claim'".to_owned(); + let before = count(&database, commands.clone()).await; + for key in ["evt-a", "evt-b"] { + let (status, answer) = bridge(&app, "acme", signal_event(key, "bridge-signal-claim", "note")).await; + assert_eq!(status, StatusCode::CONFLICT, "{key}: {answer}"); + identity_conflict(&answer, "signal_workflow", key, json!({"workflow_id":"bridge-signal-claim","signal_name":"note"})); + } + assert_eq!(count(&database, commands).await, before); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!(events(&history, "SignalReceived"), 2, "{history}"); + // The same tuple's own identity is unaffected: a fresh key is accepted. + let (status, answer) = bridge(&app, "acme", signal_event("evt-c", "bridge-signal-claim", "note")).await; + assert_eq!(status, StatusCode::ACCEPTED, "{answer}"); +} + +// A bridge start is decided under the transition lock before anything is +// written: a draining queue, then an instance another namespace holds, then +// one reserved for another type. No read outside the lock decides it, so a +// state change before the start cannot turn into another answer. + +fn start_event(key: &str, target: Value) -> Value { + json!({"action":"start_workflow","idempotency_key":key,"target":target}) +} + +async fn register_on(app: &Router, worker: &str, queue: &str, build: &str) { + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":worker,"task_queue":queue, + "runtime":"php","build_id":build,"supported_workflow_types":["echo"],"supported_activity_types":[]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +#[tokio::test] +async fn a_bridge_start_refuses_under_the_transition_lock_with_its_own_envelope() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + namespaces::create(&app, "other").await; + namespaces::start_scoped(&app, "other", "bridge-foreign").await; + let (status, body) = request(&app, "POST", "/api/workflows", json!({"workflow_id":"bridge-typed","workflow_type":"other.type", + "task_queue":"test","input":envelope(Payload::Array(vec![]))})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + let commands = "SELECT COUNT(*) FROM workflow_commands".to_owned(); + let instances = "SELECT COUNT(*) FROM workflow_instances".to_owned(); + let before = (count(&database, commands.clone()).await, count(&database, instances.clone()).await); + + let target = json!({"workflow_id":"bridge-foreign","workflow_type":"echo","task_queue":"test"}); + let (status, answer) = bridge(&app, "acme", start_event("evt-foreign", target.clone())).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{answer}"); + assert_eq!(answer, json!({"schema":"durable-workflow.v2.bridge-adapter-outcome.contract","version":1,"adapter":"acme", + "action":"start_workflow","accepted":false,"outcome":"rejected","reason":"unknown_target","idempotency_key":"evt-foreign", + "message":"Workflow instance [bridge-foreign] is unavailable.","target":target})); + + let target = json!({"workflow_id":"bridge-typed","workflow_type":"echo","task_queue":"test"}); + let (status, answer) = bridge(&app, "acme", start_event("evt-typed", target.clone())).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{answer}"); + assert_eq!(answer, json!({"schema":"durable-workflow.v2.bridge-adapter-outcome.contract","version":1,"adapter":"acme", + "action":"start_workflow","accepted":false,"outcome":"rejected","reason":"unknown_target","idempotency_key":"evt-typed", + "message":"Workflow instance [bridge-typed] is reserved for durable type [other.type] and cannot be reused for [echo].","target":target})); + + register_on(&app, "drain-worker", "drain-q", "d1").await; + let (status, body) = request(&app, "POST", "/api/task-queues/drain-q/build-ids/drain", json!({"build_id":"d1"})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, answer) = bridge(&app, "acme", start_event("evt-drain", json!({"workflow_type":"echo","task_queue":"drain-q"}))).await; + assert_eq!((status, &answer["reason"], &answer["routing_status"]), (StatusCode::UNPROCESSABLE_ENTITY, &json!("task_queue_draining"), &json!("draining")), "{answer}"); + assert_eq!((&answer["control_plane_outcome"], &answer["draining_worker_count"]), (&json!("rejected_task_queue_draining"), &json!(1)), "{answer}"); + assert_eq!(answer["message"], "Task queue [drain-q] is draining and cannot accept new workflow starts until an active worker cohort is available."); + assert_eq!((count(&database, commands).await, count(&database, instances).await), before); +} + +// A controlled interleaving with the real transition lock (Security, +// server-325 2219). The test holds the lock; a bridge start queues on it, +// then a second test-held lock, then a competitor: a start of the same id in +// another namespace, a start of it for another type, or a drain of the +// bridge's queue. Released, the bridge is granted first; the second hold +// then keeps the competitor out until a read the bridge made outside the +// lock has finished, so a start decided outside the lock (867a2d53) meets +// the competitor's state at its write. Decided in one serialized +// transaction, the bridge has started its own instance before the +// competitor runs. PostgreSQL's advisory lock and InnoDB's row lock grant +// waiters in order; SQLite's busy retry has no queue, so there the order is +// not forced and the bridge may meet either winner, but never with a raw +// answer or another type's instance. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_foreign_start_queued_behind_a_bridge_start_cannot_change_its_answer() { + change_queued_behind_a_bridge_start("namespace").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn another_types_start_queued_behind_a_bridge_start_cannot_change_its_answer() { + change_queued_behind_a_bridge_start("type").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_drain_queued_behind_a_bridge_start_cannot_change_its_answer() { + change_queued_behind_a_bridge_start("drain").await; +} + +async fn change_queued_behind_a_bridge_start(case: &'static str) { + use std::time::Duration; + let pause = || tokio::time::sleep(Duration::from_millis(200)); + { + let database = TestDatabase::new().await; + let ordered = !matches!(database, TestDatabase::Sqlite(_)); + let app = router(database.open().await.unwrap()); + namespaces::create(&app, "other").await; + register_on(&app, "queued-worker", "queued-q", "q1").await; + let workflow_id = format!("bridge-queued-{case}"); + let queue = if case == "drain" { "queued-q" } else { "test" }; + let target = json!({"workflow_id":workflow_id,"workflow_type":"echo","task_queue":queue}); + + let first = database.transition_lock().await; + let bridged = { + let (app, event) = (app.clone(), start_event(&format!("evt-queued-{case}"), target.clone())); + tokio::spawn(async move { bridge(&app, "acme", event).await }) + }; + pause().await; + assert!(!bridged.is_finished(), "{case}: the bridge start waits for the transition lock"); + let second = tokio::spawn(database.transition_lock()); + pause().await; + assert!(!second.is_finished(), "{case}: the second hold queues behind the bridge"); + let competitor = { + let (app, workflow_id) = (app.clone(), workflow_id.clone()); + tokio::spawn(async move { + let input = envelope(Payload::Array(vec![])); + match case { + "namespace" => namespaces::scoped(&app, "other", "POST", "/api/workflows", + json!({"workflow_id":workflow_id,"workflow_type":"echo","task_queue":"test","input":input})).await, + "type" => request(&app, "POST", "/api/workflows", + json!({"workflow_id":workflow_id,"workflow_type":"other.type","task_queue":"test","input":input})).await, + _ => request(&app, "POST", "/api/task-queues/queued-q/build-ids/drain", json!({"build_id":"q1"})).await, + } + }) + }; + pause().await; + assert!(!competitor.is_finished(), "{case}: the competitor queues behind the second hold"); + first.release().await; + let second = second.await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + second.release().await; + let (status, answer) = bridged.await.unwrap(); + let (competitor_status, competitor_answer) = competitor.await.unwrap(); + + let started = status == StatusCode::ACCEPTED; + if started { + assert_eq!((&answer["outcome"], &answer["workflow_id"], &answer["workflow_type"]), + (&json!("accepted"), &json!(workflow_id), &json!("echo")), "{case}: {answer}"); + let (status, shown) = request(&app, "GET", &format!("/api/workflows/{workflow_id}"), Value::Null).await; + assert_eq!((status, &shown["workflow_type"]), (StatusCode::OK, &json!("echo")), "{case}: {shown}"); + match case { + "namespace" => assert_eq!((competitor_status, &competitor_answer["reason"]), + (StatusCode::CONFLICT, &json!("workflow_id_reserved_in_namespace")), "{case}: {competitor_answer}"), + "type" => assert_eq!((competitor_status, &competitor_answer["reason"]), + (StatusCode::CONFLICT, &json!("instance_already_started")), "{case}: {competitor_answer}"), + _ => assert_eq!(competitor_status, StatusCode::OK, "{case}: {competitor_answer}"), + } + } else { + // Only where waiters are not granted in order may the competitor + // win, and then the bridge answers that state in its envelope. + assert!(!ordered, "{case}: the bridge was granted first but answered {status} {answer}"); + let message = match case { + "namespace" => format!("Workflow instance [{workflow_id}] is unavailable."), + "type" => format!("Workflow instance [{workflow_id}] is reserved for durable type [other.type] and cannot be reused for [echo]."), + _ => "Task queue [queued-q] is draining and cannot accept new workflow starts until an active worker cohort is available.".to_owned(), + }; + assert_eq!((status, &answer["schema"], &answer["outcome"], &answer["message"]), + (StatusCode::UNPROCESSABLE_ENTITY, &json!("durable-workflow.v2.bridge-adapter-outcome.contract"), &json!("rejected"), &json!(message)), + "{case}: {answer}"); + // The competitor's own success is what stands, and the bridge + // recorded nothing. + let (shown_status, shown) = request(&app, "GET", &format!("/api/workflows/{workflow_id}"), Value::Null).await; + match case { + "namespace" => { + assert_eq!((competitor_status, &competitor_answer["workflow_type"], &competitor_answer["namespace"]), + (StatusCode::CREATED, &json!("echo"), &json!("other")), "{case}: {competitor_answer}"); + assert_eq!(shown_status, StatusCode::NOT_FOUND, "{case}: the default namespace cannot see it: {shown}"); + let (status, foreign) = namespaces::scoped(&app, "other", "GET", &format!("/api/workflows/{workflow_id}"), Value::Null).await; + assert_eq!((status, &foreign["workflow_type"]), (StatusCode::OK, &json!("echo")), "{case}: {foreign}"); + } + "type" => { + assert_eq!((competitor_status, &competitor_answer["workflow_type"]), + (StatusCode::CREATED, &json!("other.type")), "{case}: {competitor_answer}"); + assert_eq!((shown_status, &shown["workflow_type"]), (StatusCode::OK, &json!("other.type")), "{case}: {shown}"); + } + _ => { + assert_eq!((competitor_status, &competitor_answer["drain_intent"]), + (StatusCode::OK, &json!("draining")), "{case}: {competitor_answer}"); + assert_eq!(shown_status, StatusCode::NOT_FOUND, "{case}: the bridge started nothing: {shown}"); + } + } + let commands = count(&database, format!("SELECT COUNT(*) FROM workflow_commands WHERE workflow_instance_id='{workflow_id}'")).await; + assert_eq!(commands, if case == "drain" { 0 } else { 1 }, "{case}: only the competitor's start command"); + } + eprintln!("queued {case}: bridge {status}, competitor {competitor_status}"); + } +} diff --git a/rust/tests/execution/worker_pressure.rs b/rust/tests/execution/worker_pressure.rs index 0ed1764c9..5e06dd0fb 100644 --- a/rust/tests/execution/worker_pressure.rs +++ b/rust/tests/execution/worker_pressure.rs @@ -4,14 +4,14 @@ use std::time::{Duration, Instant}; // An independent physical connection holds the actual native transition lock. // No injected sqlx errors, shortened production timeouts or mocked storage. -enum ShutdownWriteLock { +pub(super) enum ShutdownWriteLock { Sqlite(sqlx::Transaction<'static, sqlx::Sqlite>, sqlx::SqlitePool), Postgres(sqlx::Transaction<'static, sqlx::Postgres>, sqlx::PgPool), MySql(sqlx::Transaction<'static, sqlx::MySql>, sqlx::MySqlPool), } impl ShutdownWriteLock { - async fn release(self) { + pub(super) async fn release(self) { match self { Self::Sqlite(tx, pool) => { tx.commit().await.unwrap(); @@ -29,44 +29,60 @@ impl ShutdownWriteLock { } } +enum LockTarget { + Sqlite(String), + Postgres(PgConnectOptions), + MySql(MySqlConnectOptions), +} + impl TestDatabase { - async fn hold_shutdown_write(&self) -> ShutdownWriteLock { - match self { - Self::Sqlite(dir) => { - let pool = sqlx::sqlite::SqlitePoolOptions::new() - .max_connections(1) - .connect(&format!( - "sqlite://{}", - dir.path().join("runtime.sqlite").display() - )) - .await - .unwrap(); - let tx = pool.begin_with("BEGIN IMMEDIATE").await.unwrap(); - ShutdownWriteLock::Sqlite(tx, pool) - } - Self::Postgres { options, .. } => { - let pool = sqlx::postgres::PgPoolOptions::new() - .max_connections(1) - .connect_with(options.as_ref().clone()) - .await - .unwrap(); - let mut tx = pool.begin().await.unwrap(); - sqlx::query("SELECT pg_advisory_xact_lock(4924501473315676163)") - .execute(&mut *tx) - .await - .unwrap(); - ShutdownWriteLock::Postgres(tx, pool) - } - Self::MySql { options, .. } => { - let pool = sqlx::mysql::MySqlPoolOptions::new() - .max_connections(1) - .connect_with(options.as_ref().clone()) - .await - .unwrap(); - let mut tx = pool.begin().await.unwrap(); - sqlx::query("SELECT engine FROM dw_server_schema WHERE engine='rust-development' FOR UPDATE") - .fetch_one(&mut *tx).await.unwrap(); - ShutdownWriteLock::MySql(tx, pool) + pub(super) async fn hold_shutdown_write(&self) -> ShutdownWriteLock { + self.transition_lock().await + } + + /// The transition lock taken on its own connection, as an owned future + /// a test can spawn and queue behind other waiters. + pub(super) fn transition_lock(&self) -> impl std::future::Future + Send + 'static { + let target = match self { + Self::Sqlite(dir) => LockTarget::Sqlite(format!("sqlite://{}", dir.path().join("runtime.sqlite").display())), + Self::Postgres { options, .. } => LockTarget::Postgres(options.as_ref().clone()), + Self::MySql { options, .. } => LockTarget::MySql(options.as_ref().clone()), + }; + async move { + match target { + LockTarget::Sqlite(url) => { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect(&url) + .await + .unwrap(); + let tx = pool.begin_with("BEGIN IMMEDIATE").await.unwrap(); + ShutdownWriteLock::Sqlite(tx, pool) + } + LockTarget::Postgres(options) => { + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .unwrap(); + let mut tx = pool.begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock(4924501473315676163)") + .execute(&mut *tx) + .await + .unwrap(); + ShutdownWriteLock::Postgres(tx, pool) + } + LockTarget::MySql(options) => { + let pool = sqlx::mysql::MySqlPoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .unwrap(); + let mut tx = pool.begin().await.unwrap(); + sqlx::query("SELECT engine FROM dw_server_schema WHERE engine='rust-development' FOR UPDATE") + .fetch_one(&mut *tx).await.unwrap(); + ShutdownWriteLock::MySql(tx, pool) + } } } } diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt index e204a261d..a67da9d4a 100644 --- a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt +++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt @@ -1,7 +1,6 @@ # Published Server 2.11.1 API routes that native answers with 501 # rust_capability_not_implemented, in route-table order. Checked by # rust/tests/execution/route_coverage.rs. -POST /api/bridge-adapters/webhook/{adapter} GET /api/service-endpoints POST /api/service-endpoints GET /api/service-endpoints/{endpointName} diff --git a/scripts/conformance/server-parity/bridge-adapters-contract.mjs b/scripts/conformance/server-parity/bridge-adapters-contract.mjs new file mode 100644 index 000000000..6c66af15a --- /dev/null +++ b/scripts/conformance/server-parity/bridge-adapters-contract.mjs @@ -0,0 +1,43 @@ +import assert from 'node:assert/strict'; + +// Reviewed bridge adapter webhooks (BridgeAdapterController), recorded +// against published Server 2.11.1. The fixture's `expected` holds the whole +// normalized observation, with the probe's workflow prefix as `{id}`, run IDs +// as labels and derived workflow IDs as `{derived-}`. Every step +// compares exactly. +// +// Published defect boundaries (fixture `published_defects`): a bridge update +// skips external update admission, so an update to an SDK-worker workflow is +// rejected workflow_definition_unavailable; and a start for an id another +// namespace holds names that namespace's type (Server #777). At each +// defect's steps the published answers or the intended ones are accepted, as +// a set per defect; nothing else is relaxed. +export function checkBridgeAdapters(fixture, observation, workflowId) { + const state = observation.bridge_adapters; + const projection = {scope: 'bridge_adapters'}; + assert.ok(state && typeof state === 'object', 'actual bridge adapters observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_bridge_adapters'}, + 'embedded bridge adapters explicitly inapplicable'); + return projection; + } + const definition = fixture.bridge_adapters; + const actual = JSON.parse(JSON.stringify(state).replaceAll(workflowId, '{id}')); + for (const {published: defect} of Object.values(definition.published_defects)) { + const steps = Object.keys(defect); + let published = true; + try { + for (const step of steps) assert.deepStrictEqual(actual[step], defect[step]); + } catch { + published = false; + } + if (published) { + for (const step of steps) actual[step] = definition.expected[step]; + } + } + for (const [step, expected] of Object.entries(definition.expected)) { + assert.deepStrictEqual(actual[step], expected, `bridge adapters ${step}`); + } + assert.deepStrictEqual(Object.keys(actual).sort(), Object.keys(definition.expected).sort(), 'bridge adapters steps'); + return projection; +} diff --git a/scripts/conformance/server-parity/bridge-adapters-probe.php b/scripts/conformance/server-parity/bridge-adapters-probe.php new file mode 100644 index 000000000..215a852d2 --- /dev/null +++ b/scripts/conformance/server-parity/bridge-adapters-probe.php @@ -0,0 +1,156 @@ +createNamespace($namespace, 'bridge adapters λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new BridgeAdapterHttp(['base_uri' => rtrim($url, '/'), 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $raw = static function (string $method, string $path, ?array $body = null, bool $worker = false) use ($http, $token, $namespace): array { + $headers = array_filter(['Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + $worker ? 'X-Durable-Workflow-Protocol-Version' : 'X-Durable-Workflow-Control-Plane-Version' => $worker ? '1.20' : '2']); + $response = $http->request($method, $path, ['headers' => $headers] + ($body === null ? [] : ['json' => $body])); + $decoded = json_decode((string) $response->getBody(), true); + if (is_array($decoded)) { + unset($decoded['server_capabilities'], $decoded['control_plane'], $decoded['protocol_version']); + } + + return ['status' => $response->getStatusCode(), 'response' => $decoded]; + }; + $ok = static function (array $recorded, int $status, string $operation): array { + if ($recorded['status'] !== $status) { + throw new RuntimeException('Bridge adapters '.$operation.': '.json_encode($recorded)); + } + + return $recorded['response']; + }; + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $queue = $workflowId.'-bra-queue'; + $worker = $workflowId.'-bra-worker'; + $type = $definition['workflow_type']; + $contract = ['signals' => ['ping'], 'signal_contracts' => [['name' => 'ping', 'parameters' => []]], + 'queries' => [], 'updates' => ['bump'], 'update_contracts' => [['name' => 'bump', 'parameters' => []]], + // `bump` has no validator. Without this declaration the regular update + // route refuses it with update_validator_contract_missing (Server #747). + 'update_validators' => []]; + $capabilities = ['query_tasks', 'workflow_updates']; + $ok($raw('POST', '/api/worker/register', ['worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'python', + 'supported_workflow_types' => [$type], 'supported_activity_types' => [], + 'workflow_command_contracts' => [$type => $contract], 'capabilities' => $capabilities, + 'capability_manifest' => CapabilityManifest::forCapabilities($capabilities)], true), 201, 'registration'); + $webhook = static fn (string $adapter, array $body): array => $raw('POST', '/api/bridge-adapters/webhook/'.rawurlencode($adapter), $body); + // Run identities become labels, in order of first appearance. + $labels = []; + $label = static function (?string $runId) use (&$labels): ?string { + if ($runId === null) { + return null; + } + $labels[$runId] ??= 'run-'.(count($labels) + 1); + + return $labels[$runId]; + }; + $answer = static function (array $recorded) use ($label): array { + $body = $recorded['response']; + if (is_array($body) && array_key_exists('run_id', $body)) { + $body['run_id'] = $label($body['run_id']); + } + + return ['status' => $recorded['status'], 'response' => $body]; + }; + $adapter = 'acme.crm:v1'; + $id = $workflowId.'-bra-order'; + $correlation = ['trace' => 'λ-1']; + $state = ['applicable' => true, 'namespace' => $namespace]; + + // Refusals before any action. + $state['bad_adapter'] = $answer($webhook('bad adapter!', ['action' => 'start_workflow', 'idempotency_key' => 'k', 'target' => ['workflow_type' => $type]])); + $state['malformed'] = $answer($webhook($adapter, ['action' => 5, 'target' => 'x'])); + $state['unsupported_action'] = $answer($webhook($adapter, ['action' => 'cancel_workflow', 'idempotency_key' => 'k-0', + 'target' => ['workflow_id' => $id, 'secret' => 'not echoed'], 'correlation' => $correlation])); + + // start_workflow. + $state['start_bad_target'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-1', + 'target' => ['workflow_type' => $type, 'workflow_id' => 'bad id!', 'duplicate_policy' => 'maybe']])); + // Derived workflow IDs are global, so their keys are unique to the run. + $unknownKey = $workflowId.'-k-2'; + $derivedKey = $workflowId.'-derive-λ'; + $state['start_unknown_type'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => $unknownKey, + 'target' => ['workflow_type' => 'parity.v1.unconfigured']])); + // The probe's own workflow in the default namespace: an id reserved for + // another type, then for another namespace. + $state['start_other_type'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-6', 'input' => $empty, + 'target' => ['workflow_type' => $type, 'workflow_id' => $workflowId, 'task_queue' => $queue]])); + $state['start_other_namespace'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-7', 'input' => $empty, + 'target' => ['workflow_type' => $fixture['workflow_type'], 'workflow_id' => $workflowId, 'task_queue' => $queue]])); + $state['start'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-3', 'input' => $empty, + 'target' => ['workflow_type' => $type, 'workflow_id' => $id, 'task_queue' => $queue, 'business_key' => 'order-1'], + 'correlation' => $correlation])); + $state['start_again'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-4', 'input' => $empty, + 'target' => ['workflow_type' => $type, 'workflow_id' => $id, 'task_queue' => $queue]])); + $state['start_reject_duplicate'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => 'k-5', 'input' => $empty, + 'target' => ['workflow_type' => $type, 'workflow_id' => $id, 'task_queue' => $queue, 'duplicate_policy' => 'reject_duplicate']])); + $state['start_derived'] = $answer($webhook($adapter, ['action' => 'start_workflow', 'idempotency_key' => $derivedKey, 'input' => $empty, + 'target' => ['workflow_type' => $type, 'task_queue' => $queue]])); + + // The started run idles on a signal wait, so signals and updates apply. + $task = $ok($raw('POST', '/api/worker/workflow-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, 'timeout_seconds' => 5], true), + 200, 'workflow task')['task'] ?? []; + $state['idled'] = $raw('POST', '/api/worker/workflow-tasks/'.($task['task_id'] ?? 'missing').'/complete', [ + 'lease_owner' => $task['lease_owner'] ?? null, 'workflow_task_attempt' => $task['workflow_task_attempt'] ?? null, + 'commands' => [['type' => 'open_signal_wait', 'signal_name' => 'ping']]], true)['status']; + + // signal_workflow. + $state['signal_unknown'] = $answer($webhook($adapter, ['action' => 'signal_workflow', 'idempotency_key' => 's-0', + 'target' => ['workflow_id' => $workflowId.'-bra-missing', 'signal_name' => 'ping']])); + $state['signal_bad_target'] = $answer($webhook($adapter, ['action' => 'signal_workflow', 'idempotency_key' => 's-1', + 'target' => ['workflow_id' => $id]])); + $state['signal'] = $answer($webhook($adapter, ['action' => 'signal_workflow', 'idempotency_key' => 's-2', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'signal_name' => 'ping'], 'correlation' => $correlation])); + $state['signal_again'] = $answer($webhook($adapter, ['action' => 'signal_workflow', 'idempotency_key' => 's-2', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'signal_name' => 'ping']])); + $state['signal_other_adapter'] = $answer($webhook('other', ['action' => 'signal_workflow', 'idempotency_key' => 's-2', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'signal_name' => 'ping']])); + + // update_workflow. + $state['update_unknown'] = $answer($webhook($adapter, ['action' => 'update_workflow', 'idempotency_key' => 'u-0', + 'target' => ['workflow_id' => $workflowId.'-bra-missing', 'update_name' => 'bump']])); + $state['update_bad_target'] = $answer($webhook($adapter, ['action' => 'update_workflow', 'idempotency_key' => 'u-1', + 'target' => ['workflow_id' => $id, 'update_name' => 'bump', 'wait_for' => 'never', 'wait_timeout_seconds' => 0]])); + $state['update'] = $answer($webhook($adapter, ['action' => 'update_workflow', 'idempotency_key' => 'u-2', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'update_name' => 'bump', 'wait_for' => 'accepted']])); + $state['update_again'] = $answer($webhook($adapter, ['action' => 'update_workflow', 'idempotency_key' => 'u-2', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'update_name' => 'bump']])); + $state['update_undeclared'] = $answer($webhook($adapter, ['action' => 'update_workflow', 'idempotency_key' => 'u-3', 'input' => $empty, + 'target' => ['workflow_id' => $id, 'update_name' => 'not_declared']])); + + // The bridge's command context on the run's history. + $history = $ok($raw('GET', '/api/workflows/'.rawurlencode($id).'/runs/'.($task['run_id'] ?? 'missing').'/history'), 200, 'history'); + $state['history_event_types'] = array_column($history['events'] ?? [], 'event_type'); + + // PHP's derivation, `bridge--`, computed + // here independently: an ID derived any other way stays in the record. + $derive = static fn (string $key): string => 'bridge-acme.crm:v1-'.substr(hash('sha256', $key), 0, 32); + + return json_decode(strtr(json_encode($state, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), [ + $derive($unknownKey) => '{derived-k-2}', $derive($derivedKey) => '{derived-derive-λ}', + ]), true, 512, JSON_THROW_ON_ERROR); +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 6227b4b7a..f5a237c5c 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -13,6 +13,7 @@ import {checkLateWorkflowCompletion} from './late-workflow-completion-contract.m import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkStorageTest} from './storage-test-contract.mjs'; import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; +import {checkBridgeAdapters} from './bridge-adapters-contract.mjs'; import {checkNexusOperations} from './nexus-operations-contract.mjs'; import {checkRecordLocalActivity} from './record-local-activity-contract.mjs'; import {checkWorkflowPollRequest} from './workflow-poll-request-contract.mjs'; @@ -163,6 +164,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const searchAttributes = fixture.search_attributes ? checkSearchAttributes(fixture, observation, workflowId) : null; const storageTest = fixture.storage_test ? checkStorageTest(fixture, observation, workflowId) : null; const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null; + const bridgeAdapters = fixture.bridge_adapters ? checkBridgeAdapters(fixture, observation, workflowId) : null; const nexusOperations = fixture.nexus_operations ? checkNexusOperations(fixture, observation, workflowId) : null; const recordLocalActivity = fixture.record_local_activity ? checkRecordLocalActivity(fixture, observation, workflowId) : null; const workflowPollRequest = fixture.workflow_poll_request ? checkWorkflowPollRequest(fixture, observation, workflowId) : null; @@ -731,6 +733,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(searchAttributes ? {search_attributes: searchAttributes} : {}), ...(storageTest ? {storage_test: storageTest} : {}), ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}), + ...(bridgeAdapters ? {bridge_adapters: bridgeAdapters} : {}), ...(nexusOperations ? {nexus_operations: nexusOperations} : {}), ...(recordLocalActivity ? {record_local_activity: recordLocalActivity} : {}), ...(workflowPollRequest ? {workflow_poll_request: workflowPollRequest} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index d39f4f734..ec38d7b86 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -431,6 +431,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $searchAttributes = isset($fixture['search_attributes']) ? finishHttpSearchAttributes($client, $fixture, $workflowId, $url) : null; $storageTest = isset($fixture['storage_test']) ? finishHttpStorageTest($client, $fixture, $workflowId, $url) : null; $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null; + $bridgeAdapters = isset($fixture['bridge_adapters']) ? finishHttpBridgeAdapters($client, $fixture, $workflowId, $url) : null; $nexusOperations = isset($fixture['nexus_operations']) ? finishHttpNexusOperations($client, $fixture, $workflowId, $url) : null; $recordLocalActivity = isset($fixture['record_local_activity']) ? finishHttpRecordLocalActivity($client, $fixture, $workflowId, $url) : null; $workflowPollRequest = isset($fixture['workflow_poll_request']) ? finishHttpWorkflowPollRequest($client, $fixture, $workflowId, $url) : null; @@ -552,6 +553,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'search_attributes' => $searchAttributes, 'storage_test' => $storageTest, 'activity_lease_renewal' => $activityLeaseRenewal, + 'bridge_adapters' => $bridgeAdapters, 'nexus_operations' => $nexusOperations, 'record_local_activity' => $recordLocalActivity, 'workflow_poll_request' => $workflowPollRequest, @@ -876,6 +878,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/search-attribute-probe.php'; require __DIR__.'/storage-test-probe.php'; require __DIR__.'/activity-lease-renewal-probe.php'; + require __DIR__.'/bridge-adapters-probe.php'; require __DIR__.'/nexus-operations-probe.php'; require __DIR__.'/record-local-activity-probe.php'; require __DIR__.'/workflow-poll-request-probe.php'; @@ -1057,6 +1060,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['activity_lease_renewal'])) { $observation['activity_lease_renewal'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } + if ($mode === 'embedded' && isset($fixture['bridge_adapters'])) { + $observation['bridge_adapters'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_bridge_adapters']; + } if ($mode === 'embedded' && isset($fixture['nexus_operations'])) { $observation['nexus_operations'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_nexus_operations']; } diff --git a/tests/Feature/BridgeAdapterControllerTest.php b/tests/Feature/BridgeAdapterControllerTest.php index 443e1cc48..764b8f92b 100644 --- a/tests/Feature/BridgeAdapterControllerTest.php +++ b/tests/Feature/BridgeAdapterControllerTest.php @@ -262,6 +262,38 @@ public function test_webhook_bridge_bounds_update_wait_timeout_before_admission( ->count()); } + public function test_regular_routes_refuse_the_bridge_request_id_namespace(): void + { + Queue::fake(); + + $start = $this->withHeaders($this->apiHeaders()) + ->postJson('/api/workflows', [ + 'workflow_id' => 'wf-bridge-preclaim', + 'workflow_type' => 'tests.interactive-command-workflow', + ]); + + $start->assertCreated(); + $this->runReadyWorkflowTask((string) $start->json('run_id')); + $runId = (string) $start->json('run_id'); + + foreach ([ + ['/api/workflows/wf-bridge-preclaim/update/approve', 'dw-bridge:v1:0000'], + ["/api/workflows/wf-bridge-preclaim/runs/{$runId}/update/approve", 'DW-BRIDGE:v1:0000'], + ['/api/workflows/wf-bridge-preclaim/signal/advance', 'dw-bridge:v1:0000'], + ["/api/workflows/wf-bridge-preclaim/runs/{$runId}/signal/advance", 'dw-bridge:'], + ] as [$path, $requestId]) { + $this->withHeaders($this->apiHeaders()) + ->postJson($path, ['input' => [true, 'api'], 'request_id' => $requestId]) + ->assertStatus(422) + ->assertJsonPath('errors.request_id.0', 'The request id field format is invalid.'); + } + + $this->assertSame(0, WorkflowCommand::query() + ->where('workflow_instance_id', 'wf-bridge-preclaim') + ->whereIn('command_type', ['signal', 'update']) + ->count()); + } + public function test_webhook_bridge_rejects_json_tagged_update_before_returning_duplicate_outcome(): void { Queue::fake(); diff --git a/tests/Fixtures/ServerParity/bridge-adapters.json b/tests/Fixtures/ServerParity/bridge-adapters.json new file mode 100644 index 000000000..d1377b26a --- /dev/null +++ b/tests/Fixtures/ServerParity/bridge-adapters.json @@ -0,0 +1,588 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "bridge-adapters", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": { + "message": "bridge adapters λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "bridge adapters λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "bridge_adapters": { + "workflow_type": "parity.v1.bridge", + "empty_blob": "wwHioz3/VYAiNwwA", + "published_defects": { + "update_skips_external_admission": { + "issue": "https://github.com/durable-workflow/server/issues/747", + "summary": "BridgeAdapterController::updateWorkflow calls WorkflowControlPlane::update directly and skips ExternalWorkflowUpdateAdmission, which WorkflowController::update tries first, so a bridge update to a workflow served by an SDK worker is rejected workflow_definition_unavailable and recorded as UpdateRejected, and its replay is rejected again.", + "published": { + "update": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "workflow_definition_unavailable", + "idempotency_key": "u-2", + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "bump" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_workflow_definition_unavailable" + } + }, + "update_again": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "workflow_definition_unavailable", + "idempotency_key": "u-2", + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "bump" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_workflow_definition_unavailable" + } + }, + "history_event_types": [ + "StartAccepted", + "WorkflowStarted", + "StartAccepted", + "StartRejected", + "SignalWaitOpened", + "SignalReceived", + "SignalReceived", + "UpdateRejected", + "UpdateRejected", + "UpdateRejected" + ] + } + }, + "foreign_instance_names_type": { + "issue": "https://github.com/durable-workflow/server/issues/777", + "summary": "DefaultWorkflowControlPlane::reserveInstance checks a reserved instance's type before its namespace, so a bridge start for an id another namespace holds is refused with that namespace's durable type named, and with a different-namespace message when the types match. Native checks the namespace first and refuses a foreign instance as unavailable, naming neither.", + "published": { + "start_other_type": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-6", + "message": "Workflow instance [{id}] is reserved for durable type [parity.v1.echo] and cannot be reused for [parity.v1.bridge].", + "target": { + "workflow_id": "{id}", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue" + } + } + }, + "start_other_namespace": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-7", + "message": "Workflow instance [{id}] cannot be reused with a different namespace.", + "target": { + "workflow_id": "{id}", + "workflow_type": "parity.v1.echo", + "task_queue": "{id}-bra-queue" + } + } + } + } + } + }, + "expected": { + "applicable": true, + "namespace": "{id}-bra", + "bad_adapter": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "bad adapter!", + "accepted": false, + "outcome": "rejected", + "reason": "unsupported_routing", + "message": "Bridge adapter route is unsupported." + } + }, + "malformed": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "message": "Bridge event payload is malformed.", + "errors": { + "action": [ + "The action field must be a string." + ], + "idempotency_key": [ + "The idempotency key field is required." + ], + "target": [ + "The target field must be an array." + ] + } + } + }, + "unsupported_action": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "cancel_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unsupported_action", + "idempotency_key": "k-0", + "message": "Bridge action is unsupported.", + "target": { + "workflow_id": "{id}-bra-order" + }, + "correlation": { + "trace": "λ-1" + } + } + }, + "start_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "k-1", + "message": "Bridge start target is malformed.", + "errors": { + "workflow_id": [ + "The workflow id field format is invalid." + ], + "duplicate_policy": [ + "The selected duplicate policy is invalid." + ] + }, + "target": { + "workflow_id": "bad id!", + "workflow_type": "parity.v1.bridge" + } + } + }, + "start_unknown_type": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "{id}-k-2", + "target": { + "workflow_id": "{derived-k-2}", + "workflow_type": "parity.v1.unconfigured" + }, + "workflow_id": "{derived-k-2}", + "run_id": "run-1", + "workflow_type": "parity.v1.unconfigured", + "control_plane_outcome": "started_new" + } + }, + "start_other_type": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-6", + "message": "Workflow instance [{id}] is unavailable.", + "target": { + "workflow_id": "{id}", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue" + } + } + }, + "start_other_namespace": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-7", + "message": "Workflow instance [{id}] is unavailable.", + "target": { + "workflow_id": "{id}", + "workflow_type": "parity.v1.echo", + "task_queue": "{id}-bra-queue" + } + } + }, + "start": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "k-3", + "target": { + "workflow_id": "{id}-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue", + "business_key": "order-1" + }, + "correlation": { + "trace": "λ-1" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "started_new" + } + }, + "start_again": { + "status": 200, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "duplicate", + "reason": "duplicate_start", + "idempotency_key": "k-4", + "target": { + "workflow_id": "{id}-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "returned_existing_active" + } + }, + "start_reject_duplicate": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "instance_already_started", + "idempotency_key": "k-5", + "target": { + "workflow_id": "{id}-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "rejection_reason": "instance_already_started", + "control_plane_outcome": "rejected_duplicate" + } + }, + "start_derived": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "{id}-derive-λ", + "target": { + "workflow_id": "{derived-derive-λ}", + "workflow_type": "parity.v1.bridge", + "task_queue": "{id}-bra-queue" + }, + "workflow_id": "{derived-derive-λ}", + "run_id": "run-3", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "started_new" + } + }, + "idled": 200, + "signal_unknown": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "s-0", + "target": { + "workflow_id": "{id}-bra-missing", + "signal_name": "ping" + } + } + }, + "signal_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "s-1", + "message": "Bridge signal target is malformed.", + "errors": { + "signal_name": [ + "The signal name field is required." + ] + }, + "target": { + "workflow_id": "{id}-bra-order" + } + } + }, + "signal": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "s-2", + "target": { + "workflow_id": "{id}-bra-order", + "signal_name": "ping" + }, + "correlation": { + "trace": "λ-1" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "signal_received" + } + }, + "signal_again": { + "status": 200, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "duplicate", + "idempotency_key": "s-2", + "target": { + "workflow_id": "{id}-bra-order", + "signal_name": "ping" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "deduped_existing_command" + } + }, + "signal_other_adapter": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "other", + "action": "signal_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "s-2", + "target": { + "workflow_id": "{id}-bra-order", + "signal_name": "ping" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "signal_received" + } + }, + "update_unknown": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "u-0", + "target": { + "workflow_id": "{id}-bra-missing", + "update_name": "bump" + } + } + }, + "update_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "u-1", + "message": "Bridge update target is malformed.", + "errors": { + "wait_for": [ + "The selected wait for is invalid." + ], + "wait_timeout_seconds": [ + "The wait timeout seconds field must be at least 1." + ] + }, + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "bump" + } + } + }, + "update": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "u-2", + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "bump" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2" + } + }, + "update_again": { + "status": 200, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "duplicate", + "idempotency_key": "u-2", + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "bump" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "deduped_existing_command" + } + }, + "update_undeclared": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_update", + "idempotency_key": "u-3", + "target": { + "workflow_id": "{id}-bra-order", + "update_name": "not_declared" + }, + "workflow_id": "{id}-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_unknown_update" + } + }, + "history_event_types": [ + "StartAccepted", + "WorkflowStarted", + "StartAccepted", + "StartRejected", + "SignalWaitOpened", + "SignalReceived", + "SignalReceived", + "UpdateAccepted", + "UpdateRejected" + ] + } + }, + "contract": "Bridge adapter webhooks (BridgeAdapterController) as published Server 2.11.1 serves them, in their own namespace: the adapter name, payload and action refusals (422, the target redacted to its routing fields); start_workflow accepted (202), returned as a duplicate (200) under the default policy, rejected under reject_duplicate (422), with a workflow ID derived from the adapter and the idempotency key when none is given, and on the default queue for an unconfigured type; signal_workflow and update_workflow refused for an unknown workflow or a malformed target, accepted (202), and deduplicated (200) by adapter, action, idempotency key and target name. Run IDs are labels in order of first appearance and derived workflow IDs are checked against PHP's derivation. Two published defects are bounded: a bridge update to an SDK-worker workflow skips external update admission and is rejected workflow_definition_unavailable, and a start for an id another namespace holds names that namespace's type; for each, the published answers or the intended ones (acceptance and deduplication; a foreign instance refused as unavailable) are accepted. Embedded applications have no HTTP bridge adapters." +} diff --git a/tests/Fixtures/ServerParityExpected/bridge-adapters-observation.json b/tests/Fixtures/ServerParityExpected/bridge-adapters-observation.json new file mode 100644 index 000000000..ad43d7217 --- /dev/null +++ b/tests/Fixtures/ServerParityExpected/bridge-adapters-observation.json @@ -0,0 +1,449 @@ +{ + "applicable": true, + "namespace": "pr1-bridge-adapters-bra", + "bad_adapter": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "bad adapter!", + "accepted": false, + "outcome": "rejected", + "reason": "unsupported_routing", + "message": "Bridge adapter route is unsupported." + } + }, + "malformed": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "message": "Bridge event payload is malformed.", + "errors": { + "action": [ + "The action field must be a string." + ], + "idempotency_key": [ + "The idempotency key field is required." + ], + "target": [ + "The target field must be an array." + ] + } + } + }, + "unsupported_action": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "cancel_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unsupported_action", + "idempotency_key": "k-0", + "message": "Bridge action is unsupported.", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order" + }, + "correlation": { + "trace": "λ-1" + } + } + }, + "start_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "k-1", + "message": "Bridge start target is malformed.", + "errors": { + "workflow_id": [ + "The workflow id field format is invalid." + ], + "duplicate_policy": [ + "The selected duplicate policy is invalid." + ] + }, + "target": { + "workflow_id": "bad id!", + "workflow_type": "parity.v1.bridge" + } + } + }, + "start_unknown_type": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "pr1-bridge-adapters-k-2", + "target": { + "workflow_id": "{derived-k-2}", + "workflow_type": "parity.v1.unconfigured" + }, + "workflow_id": "{derived-k-2}", + "run_id": "run-1", + "workflow_type": "parity.v1.unconfigured", + "control_plane_outcome": "started_new" + } + }, + "start_other_type": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-6", + "message": "Workflow instance [pr1-bridge-adapters] is reserved for durable type [parity.v1.echo] and cannot be reused for [parity.v1.bridge].", + "target": { + "workflow_id": "pr1-bridge-adapters", + "workflow_type": "parity.v1.bridge", + "task_queue": "pr1-bridge-adapters-bra-queue" + } + } + }, + "start_other_namespace": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "k-7", + "message": "Workflow instance [pr1-bridge-adapters] cannot be reused with a different namespace.", + "target": { + "workflow_id": "pr1-bridge-adapters", + "workflow_type": "parity.v1.echo", + "task_queue": "pr1-bridge-adapters-bra-queue" + } + } + }, + "start": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "k-3", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "pr1-bridge-adapters-bra-queue", + "business_key": "order-1" + }, + "correlation": { + "trace": "λ-1" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "started_new" + } + }, + "start_again": { + "status": 200, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "duplicate", + "reason": "duplicate_start", + "idempotency_key": "k-4", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "pr1-bridge-adapters-bra-queue" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "returned_existing_active" + } + }, + "start_reject_duplicate": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "instance_already_started", + "idempotency_key": "k-5", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "workflow_type": "parity.v1.bridge", + "task_queue": "pr1-bridge-adapters-bra-queue" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "workflow_type": "parity.v1.bridge", + "rejection_reason": "instance_already_started", + "control_plane_outcome": "rejected_duplicate" + } + }, + "start_derived": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "start_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "pr1-bridge-adapters-derive-λ", + "target": { + "workflow_id": "{derived-derive-λ}", + "workflow_type": "parity.v1.bridge", + "task_queue": "pr1-bridge-adapters-bra-queue" + }, + "workflow_id": "{derived-derive-λ}", + "run_id": "run-3", + "workflow_type": "parity.v1.bridge", + "control_plane_outcome": "started_new" + } + }, + "idled": 200, + "signal_unknown": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "s-0", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-missing", + "signal_name": "ping" + } + } + }, + "signal_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "s-1", + "message": "Bridge signal target is malformed.", + "errors": { + "signal_name": [ + "The signal name field is required." + ] + }, + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order" + } + } + }, + "signal": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "s-2", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "signal_name": "ping" + }, + "correlation": { + "trace": "λ-1" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "signal_received" + } + }, + "signal_again": { + "status": 200, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "signal_workflow", + "accepted": false, + "outcome": "duplicate", + "idempotency_key": "s-2", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "signal_name": "ping" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "deduped_existing_command" + } + }, + "signal_other_adapter": { + "status": 202, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "other", + "action": "signal_workflow", + "accepted": true, + "outcome": "accepted", + "idempotency_key": "s-2", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "signal_name": "ping" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "signal_received" + } + }, + "update_unknown": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_target", + "idempotency_key": "u-0", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-missing", + "update_name": "bump" + } + } + }, + "update_bad_target": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "malformed_payload", + "idempotency_key": "u-1", + "message": "Bridge update target is malformed.", + "errors": { + "wait_for": [ + "The selected wait for is invalid." + ], + "wait_timeout_seconds": [ + "The wait timeout seconds field must be at least 1." + ] + }, + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "update_name": "bump" + } + } + }, + "update": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "workflow_definition_unavailable", + "idempotency_key": "u-2", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "update_name": "bump" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_workflow_definition_unavailable" + } + }, + "update_again": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "workflow_definition_unavailable", + "idempotency_key": "u-2", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "update_name": "bump" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_workflow_definition_unavailable" + } + }, + "update_undeclared": { + "status": 422, + "response": { + "schema": "durable-workflow.v2.bridge-adapter-outcome.contract", + "version": 1, + "adapter": "acme.crm:v1", + "action": "update_workflow", + "accepted": false, + "outcome": "rejected", + "reason": "unknown_update", + "idempotency_key": "u-3", + "target": { + "workflow_id": "pr1-bridge-adapters-bra-order", + "update_name": "not_declared" + }, + "workflow_id": "pr1-bridge-adapters-bra-order", + "run_id": "run-2", + "control_plane_outcome": "rejected_unknown_update" + } + }, + "history_event_types": [ + "StartAccepted", + "WorkflowStarted", + "StartAccepted", + "StartRejected", + "SignalWaitOpened", + "SignalReceived", + "SignalReceived", + "UpdateRejected", + "UpdateRejected", + "UpdateRejected" + ] +} diff --git a/tests/Unit/ServerParityBridgeAdaptersContractTest.mjs b/tests/Unit/ServerParityBridgeAdaptersContractTest.mjs new file mode 100644 index 000000000..5ca0866fe --- /dev/null +++ b/tests/Unit/ServerParityBridgeAdaptersContractTest.mjs @@ -0,0 +1,89 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkBridgeAdapters} from '../../scripts/conformance/server-parity/bridge-adapters-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/bridge-adapters.json', import.meta.url))); +// The observation published Server 2.11.1 returned for this fixture. +const recorded = JSON.parse(readFileSync( + new URL('../Fixtures/ServerParityExpected/bridge-adapters-observation.json', import.meta.url))); +const id = 'pr1-bridge-adapters'; +const observation = (change = () => {}, base = recorded) => { + const state = structuredClone(base); + change(state); + return {mode: 'http', bridge_adapters: state}; +}; +const check = state => checkBridgeAdapters(fixture, state, id); +// The intended answers: the bridge update admitted like the regular route, +// and a foreign instance refused as unavailable. +const intended = JSON.parse(JSON.stringify(fixture.bridge_adapters.expected).replaceAll('{id}', id)); + +test('the published 2.11.1 recording satisfies the contract', () => { + assert.deepStrictEqual(check(observation()), {scope: 'bridge_adapters'}); +}); + +test('the intended update acceptance and deduplication are accepted too', () => { + assert.deepStrictEqual(check(observation(() => {}, intended)), {scope: 'bridge_adapters'}); + assert.equal(intended.update.status, 202); + assert.equal(intended.update_again.response.control_plane_outcome, 'deduped_existing_command'); +}); + +test('the intended foreign-instance refusal is accepted with either update answer', () => { + const foreign = state => { + state.start_other_type = structuredClone(intended.start_other_type); + state.start_other_namespace = structuredClone(intended.start_other_namespace); + }; + assert.deepStrictEqual(check(observation(foreign)), {scope: 'bridge_adapters'}); + assert.equal(intended.start_other_type.response.message, `Workflow instance [${id}] is unavailable.`); + assert.doesNotMatch(JSON.stringify(intended.start_other_type), /parity\.v1\.echo/); +}); + +test('embedded applications are explicitly inapplicable', () => { + assert.deepStrictEqual(check({mode: 'embedded', bridge_adapters: + {applicable: false, reason: 'embedded_has_no_http_bridge_adapters'}}), {scope: 'bridge_adapters'}); +}); + +for (const [label, change, message] of [ + // Native answered 501 for the bridge route before this fixture. + ['an unimplemented route', state => { + state.start = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; + }, /bridge adapters start/], + ['only part of the published update defect', state => { + state.update = structuredClone(intended.update); + }, /bridge adapters update_again|bridge adapters history_event_types/], + ['an accepted update that is not deduplicated', state => { + Object.assign(state, structuredClone({update: intended.update, history_event_types: intended.history_event_types})); + state.update_again = structuredClone(intended.update); + }, /bridge adapters update_again/], + ['a derived workflow id from another digest', state => { + state.start_derived.response.workflow_id = 'bridge-acme.crm:v1-0000'; + }, /bridge adapters start_derived/], + ['a replayed signal that is signalled again', state => { + state.signal_again = structuredClone(state.signal); + }, /bridge adapters signal_again/], + ['a deduplication that ignores the adapter', state => { + state.signal_other_adapter = structuredClone(state.signal_again); + }, /bridge adapters signal_other_adapter/], + ['a refusal that echoes unredacted target fields', state => { + state.unsupported_action.response.target.secret = 'not echoed'; + }, /bridge adapters unsupported_action/], + ['reject_duplicate treated as use_existing', state => { + state.start_reject_duplicate = structuredClone(state.start_again); + }, /bridge adapters start_reject_duplicate/], + ['only part of the published foreign-instance defect', state => { + state.start_other_type = structuredClone(intended.start_other_type); + }, /bridge adapters start_other_namespace/], + ['a foreign instance refused with its type named', state => { + state.start_other_namespace.response.message = state.start_other_type.response.message; + }, /bridge adapters start_other_type|bridge adapters start_other_namespace/], + ['a missing step', state => { + delete state.signal_bad_target; + }, /bridge adapters signal_bad_target/], + ['an extra step', state => { + state.extra = {status: 200}; + }, /bridge adapters steps/], +]) { + test(`the contract refuses ${label}`, () => { + assert.throws(() => check(observation(change)), message); + }); +}