diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 7be89f569..d3bed7b46 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -66,6 +66,7 @@ on: - tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs - tests/Unit/ServerParityChildDescriptionsContractTest.mjs - tests/Unit/ServerParityChildRetriesContractTest.mjs + - tests/Unit/ServerParityChildVisibilityContractTest.mjs - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs - tests/Unit/ServerParityQueryArgumentsContractTest.mjs - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs @@ -1995,7 +1996,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/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.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/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityChildDescriptionsContractTest.mjs tests/Unit/ServerParityChildRetriesContractTest.mjs tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.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/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.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/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityChildDescriptionsContractTest.mjs tests/Unit/ServerParityChildRetriesContractTest.mjs tests/Unit/ServerParityChildVisibilityContractTest.mjs tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.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/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md index 77d3a1b03..c6ab5627b 100644 --- a/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md +++ b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md @@ -67,5 +67,5 @@ delivery expects one link per call and refuses it with `409 cancellation_delivery_shape_mismatch`. Security's review confirmed this from the source. A fix needs a PHP recording first: #873. - - Memos and search attributes inherited by child runs, retried or not - (a separate gap). + - Memos and search attributes inherited by child runs, retried or not: + the [child-visibility checkpoint](2026-10-29-child-visibility.md). diff --git a/docs/rust-server-port-checkpoints/2026-10-29-child-visibility.md b/docs/rust-server-port-checkpoints/2026-10-29-child-visibility.md new file mode 100644 index 000000000..1b23b2001 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-29-child-visibility.md @@ -0,0 +1,35 @@ +# October 29, memos and search attributes on child runs + +- **Before.** Native copied a run's memos and search attributes only to a + run that continued it as new. A child's first run, and a run retrying a + failed child, started with none. So their descriptions and run lists + showed an empty memo, and a visibility query for the parent's attribute + did not find the child. +- **Contract** (published Server 2.11.7, Workflow 2.15.7 + `inheritTypedVisibilityMetadata`, `MemoUpsertService::inheritFromParent` + and `SearchAttributeUpsertService::inheritFromParent`): + - a child's first run takes its parent run's memos and search + attributes; + - a retry takes those of the run it retries; + - each copy is recorded as inherited at sequence 1. + + A visibility query lists every run of the child that matches, the failed + first run and the pending retry alike. +- **Fixture.** `child-visibility`, raw worker HTTP on protocol 1.20 in a + fresh namespace with one registered keyword attribute. A parent started + with a memo and the attribute starts one child under a retry policy. The + probe records: + - the child's description; + - after a failure, the retry's description; + - the child's run list; + - a query for the attribute. +- **Native.** The continue-as-new copy is now two shared helpers, + `inheritable_memos` and `inherit_visibility` (`continuations.rs`). + `start_child` and the child retry call them right after creating the run. +- **Tests.** `a_child_and_its_retry_inherit_the_parents_memos_and_search_attributes` + (`child_retries.rs`), and the existing continue-as-new tests for the + shared copy. +- **Also here** (Security's re-review of #869). A child backoff sent as a + JSON object, which PHP reads as its values, is now checked at admission + like a list. Before, such an extreme value was refused only when the + retry started (`invalid_child_retry_policy`), and it never overflowed. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index a729cedec..fc1c4b29c 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -616,8 +616,7 @@ contract fixture before its correction. (`invalid_child_timeout`, `invalid_child_retry_policy`), as a timer delay is. Remaining: native's cooperative cancellation delivery expects one link per call, so it refuses one delivered to a retried child - (`cancellation_delivery_shape_mismatch`; #873, pending a PHP recording); - and a retried child inherits no memos or search attributes (below). + (`cancellation_delivery_shape_mismatch`; #873, pending a PHP recording). - **Trace context (Core 2.10, published in Server 2.11.1).** Core adds nullable `observability_context` columns on runs, commands and tasks. Native has no carrier: it accepts and ignores a top-level carrier on @@ -654,10 +653,14 @@ contract fixture before its correction. - **Visibility query key order.** When one command or query has several errors, native reports them in sorted key order rather than request order. The same holds for several invalid search attributes on one start. -- **Inherited memos and search attributes on child runs.** Published PHP - copies the parent's memos and search attributes to a child run and to a - retried child (`inheritTypedVisibilityMetadata`). Native copies both only - on continue-as-new. +- **Inherited memos and search attributes on child runs** (October 29, + closed). Published PHP copies the parent's memos and search attributes to + a child run, and the failed run's to a retried child + (`inheritTypedVisibilityMetadata`). Native copied both only on + continue-as-new; the same copy now runs for a child's first run and a + retry. The `child-visibility` fixture records it against published + 2.11.7 + ([checkpoint](rust-server-port-checkpoints/2026-10-29-child-visibility.md)). - **Search attribute datetime forms.** `Carbon::parse` accepts relative and free-form dates. Native accepts RFC 3339, and a date or date and time without an offset read as UTC, and refuses other forms as PHP refuses an diff --git a/rust/src/runtime/child_retries.rs b/rust/src/runtime/child_retries.rs index 6f5b48680..d909aaf07 100644 --- a/rust/src/runtime/child_retries.rs +++ b/rust/src/runtime/child_retries.rs @@ -130,7 +130,7 @@ fn error_types(value: &Value) -> Vec { } /// A PHP array's values: a list, or an object's values in order. -fn list(value: &Value) -> Option + '_>> { +pub(super) fn list(value: &Value) -> Option + '_>> { match value { Value::Array(entries) => Some(Box::new(entries.iter())), Value::Object(entries) => Some(Box::new(entries.values())), diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index 9062a738a..1dccfaa1a 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -69,8 +69,9 @@ pub(super) fn validate_child(command: &Value) -> Result<()> { return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_timeout")); } } - if let Some(backoff) = command["retry_policy"]["backoff_seconds"].as_array() - && backoff.iter().any(|entry| entry.as_i64().is_some_and(|seconds| super::child_retries::after(now, seconds).is_none())) + // PHP reads a backoff object's values as a list (`array_values`). + if let Some(mut backoff) = super::child_retries::list(&command["retry_policy"]["backoff_seconds"]) + && backoff.any(|entry| entry.as_i64().is_some_and(|seconds| super::child_retries::after(now, seconds).is_none())) { return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_retry_policy")); } @@ -312,6 +313,10 @@ where .bind(&child_run).bind(&child_instance).bind(namespace.as_deref()).bind(workflow_type).bind(workflow_type).bind(&arguments).bind(&queue).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(compatibility.as_deref()) // Timeouts are at least one second; zero binds "no timeout". .bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)).execute(&mut **tx).await?; + // `inheritTypedVisibilityMetadata`: the child takes its parent's memos + // and search attributes. + let memos = Self::inheritable_memos(tx, &parent_run).await?; + Self::inherit_visibility(tx, memos, &parent_run, &child_run, &child_instance, started_at).await?; Self::query("INSERT INTO workflow_child_calls(parent_workflow_run_id,parent_workflow_instance_id,sequence,child_workflow_type,child_workflow_class,resolved_child_instance_id,resolved_child_run_id,parent_close_policy,cancellation_propagation,queue,status,scheduled_at,started_at,arguments,metadata,created_at,updated_at,retry_policy,timeout_policy) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,($9<>'abandon'),$10,'started',$11,$12,$13,$14,$15,$16,$17,$18)") .bind(&parent_run).bind(&parent_instance).bind(sequence).bind(workflow_type).bind(workflow_type).bind(&child_instance).bind(&child_run).bind(&policies.parent_close).bind(&policies.cancellation).bind(&queue).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)) .bind(DB::document(&json!({"payload":arguments}))).bind(DB::document(&json!({"child_call_id":call,"attempt_count":1,"cancellation_policy":policies.cancellation}))).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)) @@ -609,6 +614,9 @@ where .bind(&retry_run).bind(run_number).bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)) .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(&failed_run) .execute(&mut **tx).await?; + // The retry takes the failed run's memos and search attributes. + let memos = Self::inheritable_memos(tx, &failed_run).await?; + Self::inherit_visibility(tx, memos, &failed_run, &retry_run, &instance, at).await?; // Each placeholder appears once, in order, as MySQL's templates require. Self::query("UPDATE workflow_instances SET current_run_id=$1,run_count=CASE WHEN run_count>$2 THEN run_count ELSE $3 END,updated_at=$4 WHERE id=$5") .bind(&retry_run).bind(run_number).bind(run_number).bind(DB::bind_time(at)).bind(&instance).execute(&mut **tx).await?; diff --git a/rust/src/runtime/continuations.rs b/rust/src/runtime/continuations.rs index 1e66a2d3f..e35690e12 100644 --- a/rust/src/runtime/continuations.rs +++ b/rust/src/runtime/continuations.rs @@ -10,6 +10,7 @@ use super::{ wire, }; use axum::http::StatusCode; +use chrono::{DateTime, Utc}; use serde_json::{Value, json}; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; @@ -29,6 +30,9 @@ pub(super) fn validate(command: &Value) -> Result<()> { Ok(()) } +/// A memo key, its stored value and its current portable value. +pub(super) type InheritedMemo = (String, Value, Option); + impl Store where for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, @@ -85,6 +89,51 @@ where Ok(json!({"workflow_id":workflow,"run_count":runs.len(),"runs":runs})) } + /// The memos `MemoUpsertService::inheritFromParent` copies from a run: + /// each key and value, with its portable value when that is current. + pub(super) async fn inheritable_memos(tx: &mut Transaction<'_, DB>, from: &str) -> Result> { + let rows = Self::query(DB::MEMO_READ).bind(from).fetch_all(&mut **tx).await?; + if rows.len() > super::memos::MAX_ENTRIES { + return Err(refuse(StatusCode::CONFLICT, "workflow_memo_validation_failed")); + } + let mut memos = Vec::with_capacity(rows.len()); + for memo in rows { + let portable = if DB::optional_number(&memo, "portable_value_sequence")? + == Some(DB::number(&memo, "upserted_at_sequence")?) + { + DB::optional_document_row(&memo, "portable_value")? + .filter(|value| value.is_object() || value.is_array()) + } else { + None + }; + memos.push((DB::string(&memo, "key")?, DB::document_row(&memo, "value")?, portable)); + } + Ok(memos) + } + + /// `inheritTypedVisibilityMetadata`: a new run takes these memos and + /// the source run's search attributes, as inherited at sequence 1. + /// Continue-as-new, a child's first run and a retried child's run do. + pub(super) async fn inherit_visibility( + tx: &mut Transaction<'_, DB>, + memos: Vec, + from: &str, + to: &str, + instance: &str, + at: DateTime, + ) -> Result<()> { + for (key, projection, portable) in memos { + Self::query(DB::MEMO_INHERIT_INSERT) + .bind(to).bind(instance).bind(key).bind(DB::document(&projection)) + .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(portable.as_ref().map(DB::document)) + .execute(&mut **tx).await?; + } + Self::query(DB::SEARCH_ATTRIBUTE_INHERIT) + .bind(to).bind(instance).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(from) + .execute(&mut **tx).await?; + Ok(()) + } + pub(super) async fn continue_run( &self, tx: &mut Transaction<'_, DB>, @@ -123,32 +172,7 @@ where // resolves no call to that binding, and takeover refuses an open one. // A call linked to this run through an `activity_execution` binding // stays with the predecessor, as in Core. - let rows = Self::query(DB::MEMO_READ) - .bind(&old) - .fetch_all(&mut **tx) - .await?; - if rows.len() > super::memos::MAX_ENTRIES { - return Err(refuse( - StatusCode::CONFLICT, - "workflow_memo_validation_failed", - )); - } - let mut memos = Vec::with_capacity(rows.len()); - for memo in rows { - let portable = if DB::optional_number(&memo, "portable_value_sequence")? - == Some(DB::number(&memo, "upserted_at_sequence")?) - { - DB::optional_document_row(&memo, "portable_value")? - .filter(|value| value.is_object() || value.is_array()) - } else { - None - }; - memos.push(( - DB::string(&memo, "key")?, - DB::document_row(&memo, "value")?, - portable, - )); - } + let memos = Self::inheritable_memos(tx, &old).await?; // Validate the selected typed authority off the executor before any // successor mutation. Stale portable bytes retain the legacy fallback. super::memos::read_stored(self.memo_codec.clone(), memos.clone()).await?; @@ -191,27 +215,7 @@ where .bind(DB::bind_time(started)).bind(DB::bind_time(started)) .bind(DB::bind_time(started)).bind(DB::bind_time(started)).bind(&old) .execute(&mut **tx).await?; - for (key, projection, portable) in memos { - Self::query(DB::MEMO_INHERIT_INSERT) - .bind(&next) - .bind(&workflow) - .bind(key) - .bind(DB::document(&projection)) - .bind(DB::bind_time(started)) - .bind(DB::bind_time(started)) - .bind(portable.as_ref().map(DB::document)) - .execute(&mut **tx) - .await?; - } - // `SearchAttributeUpsertService::inheritFromParent` at sequence 1. - Self::query(DB::SEARCH_ATTRIBUTE_INHERIT) - .bind(&next) - .bind(&workflow) - .bind(DB::bind_time(started)) - .bind(DB::bind_time(started)) - .bind(&old) - .execute(&mut **tx) - .await?; + Self::inherit_visibility(tx, memos, &old, &next, &workflow, started).await?; Self::query("UPDATE workflow_instances SET current_run_id=$1,run_count=$2,workflow_class=$3,updated_at=$4 WHERE id=$5") .bind(&next).bind(number).bind(&class).bind(DB::bind_time(started)).bind(&workflow).execute(&mut **tx).await?; Self::query("INSERT INTO workflow_links(id,link_type,sequence,parent_workflow_instance_id,parent_workflow_run_id,child_workflow_instance_id,child_workflow_run_id,is_primary_parent,parent_close_policy,created_at,updated_at) VALUES ($1,'continue_as_new',$2,$3,$4,$5,$6,TRUE,'abandon',$7,$8)") diff --git a/rust/tests/execution/child_retries.rs b/rust/tests/execution/child_retries.rs index 179c63fc9..75c4c45f0 100644 --- a/rust/tests/execution/child_retries.rs +++ b/rust/tests/execution/child_retries.rs @@ -163,6 +163,7 @@ async fn a_child_deadline_or_backoff_too_far_away_is_refused_not_overflowed() { (json!({"execution_timeout_seconds": i64::MAX}), "invalid_child_timeout"), (json!({"run_timeout_seconds": i64::MAX}), "invalid_child_timeout"), (json!({"retry_policy": {"max_attempts": 2, "backoff_seconds": [1, i64::MAX]}}), "invalid_child_retry_policy"), + (json!({"retry_policy": {"max_attempts": 2, "backoff_seconds": {"first": i64::MAX}}}), "invalid_child_retry_policy"), ] { let (status, body) = finish_task(&app, &task, json!([child(options.clone())])).await; assert_eq!((status, &body["reason"]), (StatusCode::UNPROCESSABLE_ENTITY, &json!(reason)), "{options}: {body}"); @@ -173,3 +174,36 @@ async fn a_child_deadline_or_backoff_too_far_away_is_refused_not_overflowed() { runtime.close().await; database.remove().await; } + +#[tokio::test] +async fn a_child_and_its_retry_inherit_the_parents_memos_and_search_attributes() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (status, body) = request(&app, "POST", "/api/search-attributes", json!({"name":"Region","type":"keyword"})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + register(&app, "worker", json!(["echo", "child"]), json!([])).await; + let (status, parent) = request(&app, "POST", "/api/workflows", json!({"workflow_id":"inheriting","workflow_type":"echo", + "task_queue":"test","input":envelope(Payload::Array(vec![])),"memo":{"team":"ops","count":3}, + "search_attributes":{"Region":"eu"}})).await; + assert_eq!(status, StatusCode::CREATED, "{parent}"); + let task = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &task, json!([child(json!({"retry_policy":{"max_attempts":2}}))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let events = child_events(&run_history(&app, &json!("inheriting"), &parent["run_id"]).await); + let instance = events[0]["payload"]["child_workflow_instance_id"].as_str().unwrap().to_owned(); + let expected = (json!({"count":3,"team":"ops"}), json!({"Region":"eu"})); + let (_, described) = request(&app, "GET", &format!("/api/workflows/{instance}"), Value::Null).await; + assert_eq!((described["memo"].clone(), described["search_attributes"].clone()), expected, "{described}"); + let first = child_task(&app).await; + let (status, body) = finish_task(&app, &first, fail(false)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (_, described) = request(&app, "GET", &format!("/api/workflows/{instance}"), Value::Null).await; + assert_eq!(described["run_number"], 2, "{described}"); + assert_eq!((described["memo"].clone(), described["search_attributes"].clone()), expected, "{described}"); + let (_, listed) = request(&app, "GET", &format!("/api/workflows?query={}", "Region%20%3D%20%22eu%22"), Value::Null).await; + let ids: Vec<_> = listed["workflows"].as_array().unwrap().iter().map(|row| row["workflow_id"].clone()).collect(); + assert!(ids.contains(&json!(instance)), "{listed}"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/child-visibility-contract.mjs b/scripts/conformance/server-parity/child-visibility-contract.mjs new file mode 100644 index 000000000..cb4c0080c --- /dev/null +++ b/scripts/conformance/server-parity/child-visibility-contract.mjs @@ -0,0 +1,18 @@ +import assert from 'node:assert/strict'; + +// Reviewed child memos and search attributes (child-visibility-probe.php), +// recorded against published Server 2.11.7. The fixture's `expected` holds +// the whole observation. +export function checkChildVisibility(fixture, observation) { + const definition = fixture.child_visibility; + const state = observation.child_visibility; + const projection = {scope: 'child_visibility'}; + assert.ok(state && typeof state === 'object', 'actual child visibility observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded child visibility explicitly inapplicable'); + return projection; + } + assert.deepStrictEqual(state, {applicable: true, ...definition.expected}, 'child visibility'); + return projection; +} diff --git a/scripts/conformance/server-parity/child-visibility-probe.php b/scripts/conformance/server-parity/child-visibility-probe.php new file mode 100644 index 000000000..8eed83ca6 --- /dev/null +++ b/scripts/conformance/server-parity/child-visibility-probe.php @@ -0,0 +1,116 @@ +createNamespace($namespace, 'child visibility λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new ChildVisibilityHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $headers = array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => '1.20', + ]); + $raw = static function (string $method, string $path, ?array $body = null) use ($http, $headers): array { + $response = $http->request($method, $path, ['headers' => $headers, + ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $ok = static function (array $answer, int $status, string $step): array { + if ($answer['status'] !== $status) { + throw new RuntimeException("Child visibility {$step}: ".json_encode($answer)); + } + + return $answer['response']; + }; + foreach ($definition['definitions'] as [$name, $type]) { + $ok($raw('POST', 'search-attributes', ['name' => $name, 'type' => $type]), 201, 'definition'); + } + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $id = $workflowId.'-cv-parent'; + $queue = $id.'-queue'; + $worker = $id.'-worker'; + $ok($raw('POST', 'worker/register', ['worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'php', + 'sdk_version' => 'raw-http', 'supported_workflow_types' => [$definition['workflow_type'], $definition['child_type']], + 'supported_activity_types' => [], + 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http'])]), 201, 'registration'); + $started = $ok($raw('POST', 'workflows', ['workflow_id' => $id, 'workflow_type' => $definition['workflow_type'], + 'task_queue' => $queue, 'input' => $empty, 'memo' => $definition['memo'], + 'search_attributes' => $definition['search_attributes']]), 201, 'start'); + $poll = static function (string $type) use ($raw, $worker, $queue): array { + for ($attempt = 0; $attempt < 10; $attempt++) { + $task = $raw('POST', 'worker/workflow-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, + 'timeout_seconds' => 1])['response']['task'] ?? null; + if (is_array($task)) { + if (($task['workflow_type'] ?? null) !== $type) { + throw new RuntimeException("Child visibility: expected a {$type} task"); + } + + return $task; + } + } + throw new RuntimeException("Child visibility: no {$type} task"); + }; + $complete = static fn (array $task, array $commands): array => $raw('POST', + 'worker/workflow-tasks/'.$task['task_id'].'/complete', ['lease_owner' => $task['lease_owner'], + 'workflow_task_attempt' => $task['workflow_task_attempt'], 'commands' => $commands]); + $ok($complete($poll($definition['workflow_type']), [['type' => 'start_child_workflow', + 'workflow_type' => $definition['child_type'], 'arguments' => $empty, + 'retry_policy' => ['max_attempts' => 2]]]), 200, 'child start'); + $events = $ok($raw('GET', 'workflows/'.rawurlencode($id).'/runs/'.$started['run_id'].'/history'), 200, 'history')['events'] ?? []; + $child = null; + foreach ($events as $event) { + if ($event['event_type'] === 'ChildWorkflowScheduled') { + $child = $event['payload']['child_workflow_instance_id'] ?? null; + } + } + if (! is_string($child)) { + throw new RuntimeException('Child visibility: no child instance'); + } + $described = static function () use ($raw, $ok, $child): array { + $run = $ok($raw('GET', 'workflows/'.rawurlencode($child)), 200, 'child description'); + + return ['run_number' => $run['run_number'] ?? null, 'memo' => $run['memo'] ?? 'absent', + 'search_attributes' => $run['search_attributes'] ?? 'absent']; + }; + $state = ['applicable' => true, 'child' => $described()]; + $ok($complete($poll($definition['child_type']), [['type' => 'fail_workflow', 'message' => 'first attempt', + 'exception_type' => 'Boom']]), 200, 'child failure'); + $state['retry'] = $described(); + $runs = $ok($raw('GET', 'workflows/'.rawurlencode($child).'/runs'), 200, 'child runs')['runs'] ?? []; + usort($runs, static fn (array $a, array $b): int => ($a['run_number'] ?? 0) <=> ($b['run_number'] ?? 0)); + $state['runs'] = array_map(static fn (array $run): array => ['run_number' => $run['run_number'] ?? null, + 'memo' => $run['memo'] ?? 'absent'], $runs); + $listed = $ok($raw('GET', 'workflows?query='.rawurlencode($definition['query'])), 200, 'query')['workflows'] ?? []; + $names = [$id => 'parent', $child => 'child']; + $rows = array_map(static fn (array $row): array => ['workflow' => $names[$row['workflow_id'] ?? ''] ?? 'other', + 'status' => $row['status'] ?? null, 'search_attributes' => $row['search_attributes'] ?? 'absent'], $listed); + usort($rows, static fn (array $a, array $b): int => [$a['workflow'], $a['status']] <=> [$b['workflow'], $b['status']]); + $state['listed'] = $rows; + // End the child, then the parent, so nothing stays for later fixtures. + $ok($complete($poll($definition['child_type']), [['type' => 'complete_workflow', 'result' => $empty]]), 200, 'child completion'); + $ok($complete($poll($definition['workflow_type']), [['type' => 'complete_workflow', 'result' => $empty]]), 200, 'parent completion'); + + return $state; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 8b083a34b..a87f10df8 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -20,6 +20,7 @@ import {checkManyUpdates} from './many-updates-contract.mjs'; import {checkWorkerErrorBodies} from './worker-error-bodies-contract.mjs'; import {checkChildDescriptions} from './child-descriptions-contract.mjs'; import {checkChildRetries} from './child-retries-contract.mjs'; +import {checkChildVisibility} from './child-visibility-contract.mjs'; import {checkStartSearchAttributes} from './start-search-attributes-contract.mjs'; import {checkQueryArgumentRouting} from './query-arguments-contract.mjs'; import {checkQueryContractArguments} from './query-contract-arguments-contract.mjs'; @@ -185,6 +186,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const workerErrorBodies = fixture.worker_error_bodies ? checkWorkerErrorBodies(fixture, observation) : null; const childDescriptions = fixture.child_descriptions ? checkChildDescriptions(fixture, observation) : null; const childRetries = fixture.child_retries ? checkChildRetries(fixture, observation) : null; + const childVisibility = fixture.child_visibility ? checkChildVisibility(fixture, observation) : null; const startSearchAttributes = fixture.start_search_attributes ? checkStartSearchAttributes(fixture, observation) : null; const queryArguments = fixture.query_argument_routing ? checkQueryArgumentRouting(fixture, observation) : null; const queryContractArguments = fixture.query_contract_arguments ? checkQueryContractArguments(fixture, observation) : null; @@ -768,6 +770,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(workerErrorBodies ? {worker_error_bodies: workerErrorBodies} : {}), ...(childDescriptions ? {child_descriptions: childDescriptions} : {}), ...(childRetries ? {child_retries: childRetries} : {}), + ...(childVisibility ? {child_visibility: childVisibility} : {}), ...(startSearchAttributes ? {start_search_attributes: startSearchAttributes} : {}), ...(queryArguments ? {query_argument_routing: queryArguments} : {}), ...(queryContractArguments ? {query_contract_arguments: queryContractArguments} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 91ffb46b1..bf0f1d719 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -438,6 +438,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $workerErrorBodies = isset($fixture['worker_error_bodies']) ? finishHttpWorkerErrorBodies($client, $fixture, $workflowId, $url) : null; $childDescriptions = isset($fixture['child_descriptions']) ? finishHttpChildDescriptions($client, $fixture, $workflowId, $url) : null; $childRetries = isset($fixture['child_retries']) ? finishHttpChildRetries($client, $fixture, $workflowId, $url) : null; + $childVisibility = isset($fixture['child_visibility']) ? finishHttpChildVisibility($client, $fixture, $workflowId, $url) : null; $startSearchAttributes = isset($fixture['start_search_attributes']) ? finishHttpStartSearchAttributes($client, $fixture, $workflowId, $url) : null; $queryArgumentRouting = isset($fixture['query_argument_routing']) ? finishHttpQueryArgumentRouting($client, $fixture, $workflowId, $url) : null; $queryContractArguments = isset($fixture['query_contract_arguments']) ? finishHttpQueryContractArguments($client, $fixture, $workflowId, $url) : null; @@ -574,6 +575,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'worker_error_bodies' => $workerErrorBodies, 'child_descriptions' => $childDescriptions, 'child_retries' => $childRetries, + 'child_visibility' => $childVisibility, 'start_search_attributes' => $startSearchAttributes, 'query_argument_routing' => $queryArgumentRouting, 'query_contract_arguments' => $queryContractArguments, @@ -913,6 +915,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/worker-error-bodies-probe.php'; require __DIR__.'/child-descriptions-probe.php'; require __DIR__.'/child-retries-probe.php'; + require __DIR__.'/child-visibility-probe.php'; require __DIR__.'/start-search-attributes-probe.php'; require __DIR__.'/query-arguments-probe.php'; require __DIR__.'/query-contract-arguments-probe.php'; @@ -1120,6 +1123,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['child_retries'])) { $observation['child_retries'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } + if ($mode === 'embedded' && isset($fixture['child_visibility'])) { + $observation['child_visibility'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; + } if ($mode === 'embedded' && isset($fixture['start_search_attributes'])) { $observation['start_search_attributes'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_search_attribute_registry']; } diff --git a/tests/Fixtures/ServerParity/child-visibility.json b/tests/Fixtures/ServerParity/child-visibility.json new file mode 100644 index 000000000..63f55a85f --- /dev/null +++ b/tests/Fixtures/ServerParity/child-visibility.json @@ -0,0 +1,115 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "child-visibility", + "review_state": "reviewed", + "contract": "A child run's memos and search attributes (published Server 2.11.7 with Workflow 2.15.7 inheritTypedVisibilityMetadata), over raw worker HTTP on protocol 1.20 in a fresh namespace with one registered keyword attribute. A parent started with a memo and a search attribute starts one child under a retry policy. The child's first run, and the run that retries it after a failure, carry the parent's memo and search attribute in their descriptions and run list, and a visibility query for the attribute finds the child. Embedded applications have no worker protocol.", + "workflow_type": "parity.v1.echo", + "input": { + "message": "child visibility λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "child visibility λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "child_visibility": { + "workflow_type": "parity.v1.child_visibility", + "child_type": "parity.v1.child_visibility_child", + "empty_blob": "wwHioz3/VYAiNwwA", + "definitions": [ + [ + "Region", + "keyword" + ] + ], + "memo": { + "team": "ops λ", + "count": 3 + }, + "search_attributes": { + "Region": "eu" + }, + "query": "Region = \"eu\"", + "expected": { + "child": { + "run_number": 1, + "memo": { + "count": 3, + "team": "ops λ" + }, + "search_attributes": { + "Region": "eu" + } + }, + "retry": { + "run_number": 2, + "memo": { + "count": 3, + "team": "ops λ" + }, + "search_attributes": { + "Region": "eu" + } + }, + "runs": [ + { + "run_number": 1, + "memo": { + "count": 3, + "team": "ops λ" + } + }, + { + "run_number": 2, + "memo": { + "count": 3, + "team": "ops λ" + } + } + ], + "listed": [ + { + "workflow": "child", + "status": "failed", + "search_attributes": { + "Region": "eu" + } + }, + { + "workflow": "child", + "status": "pending", + "search_attributes": { + "Region": "eu" + } + }, + { + "workflow": "parent", + "status": "waiting", + "search_attributes": { + "Region": "eu" + } + } + ] + } + } +} diff --git a/tests/Unit/ServerParityChildVisibilityContractTest.mjs b/tests/Unit/ServerParityChildVisibilityContractTest.mjs new file mode 100644 index 000000000..3610dad32 --- /dev/null +++ b/tests/Unit/ServerParityChildVisibilityContractTest.mjs @@ -0,0 +1,36 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkChildVisibility} from '../../scripts/conformance/server-parity/child-visibility-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/child-visibility.json', import.meta.url), 'utf8')); +const expected = fixture.child_visibility.expected; +const observation = () => structuredClone({mode: 'http', child_visibility: {applicable: true, ...expected}}); + +test('the published answers pass', () => { + assert.deepStrictEqual(checkChildVisibility(fixture, observation()), {scope: 'child_visibility'}); +}); + +test('embedded applications have no worker protocol', () => { + assert.deepStrictEqual(checkChildVisibility(fixture, + {mode: 'embedded', child_visibility: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'child_visibility'}); +}); + +test('the child and its retry carry the parent memo and search attribute', () => { + const {memo, search_attributes: attributes} = fixture.child_visibility; + for (const run of [expected.child, expected.retry]) { + assert.deepStrictEqual([run.memo, run.search_attributes], [memo, attributes]); + } + assert.deepStrictEqual(expected.runs.map(run => run.memo), [memo, memo]); + assert.deepStrictEqual(expected.listed.map(row => row.workflow), ['child', 'child', 'parent']); +}); + +for (const [name, mutate] of [ + ['a child without the parent memo', state => { state.child.memo = []; }], + ['a retry without the search attribute', state => { state.retry.search_attributes = []; }], + ['a child missing from the query', state => { state.listed = state.listed.filter(row => row.workflow !== 'child'); }], +]) test(`the contract rejects ${name}`, () => { + const raw = observation(); mutate(raw.child_visibility); + assert.throws(() => checkChildVisibility(fixture, raw)); +});