diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 9261b3d94..da794a8f5 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -29,6 +29,7 @@ on: - tests/Unit/ServerParitySdkDirectionContractTest.mjs - tests/Unit/ServerParitySdkChildDirectionContractTest.mjs - tests/Unit/ServerParitySdkTimerAuthorContractTest.mjs + - tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs - tests/Support/ServerParityAdmissionObservation.mjs - tests/Support/ServerParityVisibilityObservation.mjs - rust/** @@ -61,6 +62,7 @@ on: - tests/Unit/ServerParitySdkDirectionContractTest.mjs - tests/Unit/ServerParitySdkChildDirectionContractTest.mjs - tests/Unit/ServerParitySdkTimerAuthorContractTest.mjs + - tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs - tests/Support/ServerParityAdmissionObservation.mjs - tests/Support/ServerParityVisibilityObservation.mjs - rust/** @@ -376,6 +378,8 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-profile.sh mysql http://parity-native-a:8080 child docker stop --time 30 "$RESOURCE_SCOPE-sdk-child-directions-server" bash scripts/conformance/server-parity/sdk-direction-profile.sh mysql http://parity-native-a:8080 timer + docker stop --time 30 "$RESOURCE_SCOPE-sdk-timers-server" + bash scripts/conformance/server-parity/sdk-direction-profile.sh mysql http://parity-native-a:8080 signal docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint sh \ -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \ -e DW_PARITY_TOKEN=parity-test-token "$image" -ec ' @@ -497,6 +501,7 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-cleanup.sh mysql bash scripts/conformance/server-parity/sdk-direction-cleanup.sh mysql child bash scripts/conformance/server-parity/sdk-direction-cleanup.sh mysql timer + bash scripts/conformance/server-parity/sdk-direction-cleanup.sh mysql signal docker inspect --format '{{json .State}}' "$RESOURCE_SCOPE-db" > parity-evidence/database-state.json 2>&1 || true docker inspect --format '{{json .HostConfig.Memory}}' "$RESOURCE_SCOPE-db" > parity-evidence/database-memory-limit.json 2>&1 || true for profile in fence pressure; do @@ -733,6 +738,8 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-profile.sh pgsql http://parity-native-a:8080 child docker stop --time 30 "$RESOURCE_SCOPE-sdk-child-directions-server" bash scripts/conformance/server-parity/sdk-direction-profile.sh pgsql http://parity-native-a:8080 timer + docker stop --time 30 "$RESOURCE_SCOPE-sdk-timers-server" + bash scripts/conformance/server-parity/sdk-direction-profile.sh pgsql http://parity-native-a:8080 signal docker run --rm --user=1000:1000 --network "$RESOURCE_SCOPE" --entrypoint sh \ -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \ -e DW_PARITY_TOKEN=parity-test-token "$image" -ec ' @@ -886,6 +893,7 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-cleanup.sh pgsql bash scripts/conformance/server-parity/sdk-direction-cleanup.sh pgsql child bash scripts/conformance/server-parity/sdk-direction-cleanup.sh pgsql timer + bash scripts/conformance/server-parity/sdk-direction-cleanup.sh pgsql signal docker inspect --format '{{json .State}}' "$RESOURCE_SCOPE-db" > parity-evidence/database-state.json 2>&1 || true docker inspect --format '{{json .HostConfig.Memory}}' "$RESOURCE_SCOPE-db" > parity-evidence/database-memory-limit.json 2>&1 || true for profile in fence pressure; do @@ -1063,6 +1071,8 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-profile.sh sqlite http://parity-rust:8080 child docker stop --time 30 "$RESOURCE_SCOPE-sdk-child-directions-server" bash scripts/conformance/server-parity/sdk-direction-profile.sh sqlite http://parity-rust:8080 timer + docker stop --time 30 "$RESOURCE_SCOPE-sdk-timers-server" + bash scripts/conformance/server-parity/sdk-direction-profile.sh sqlite http://parity-rust:8080 signal docker run --rm --user=1000:1000 --entrypoint php --network "$RESOURCE_SCOPE" \ -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \ -e DW_PARITY_TOKEN=parity-test-token "$image" scripts/conformance/server-parity/restart.php \ @@ -1155,6 +1165,7 @@ jobs: bash scripts/conformance/server-parity/sdk-direction-cleanup.sh sqlite bash scripts/conformance/server-parity/sdk-direction-cleanup.sh sqlite child bash scripts/conformance/server-parity/sdk-direction-cleanup.sh sqlite timer + bash scripts/conformance/server-parity/sdk-direction-cleanup.sh sqlite signal for profile in fence pressure; do for target in php native; do docker logs "$RESOURCE_SCOPE-$profile-$target" > "parity-evidence/$profile-$target.log" 2>&1 || true diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 4b4603bfd..e2fa3fceb 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -53,7 +53,7 @@ and [completed main qualification](https://github.com/durable-workflow/server/pu are tracked in #408. This qualifies one bounded child scenario in nine language directions; it does not establish nine capability families or performance. -The next additive profile defines normal published SDK timer authoring in PHP, +The additive timer profile qualifies normal published SDK authoring in PHP, Python and Rust. Each requests 5.25 seconds of durable sleep, rounded to six whole seconds, then cold-replays the original timer and run with a distinct author process. The [reviewed next contract](https://github.com/durable-workflow/server/issues/325#issuecomment-6112280492) @@ -64,9 +64,83 @@ dedicated reader guards and isolated profiles are integrated into all six database configurations. The existing SDK adapter build is reused. Local reader guards pass, including corrupt original deadlines, stale receipts, rescheduling, substituted Avro frames and one nanosecond early firing. -Qualification requires all six database configurations and inspection of the -retained runtime records and phase wire evidence; [#411](https://github.com/durable-workflow/server/pull/411) -tracks the execution results. +[#411](https://github.com/durable-workflow/server/pull/411) is merged with all six +candidate database configurations, normal PHP/source and boundaries passing. +The independently inspected current SQLite artifact passes nine timer +observations, all twelve full phase wire sidecars and all 55 original ZIP +member/archive roundtrip hashes. The MariaDB artifact separately passes nine +observations, 72 native execution tests and TLS checks with inspected 512 MiB +memory/swap caps. Candidate, actual PR runner and merged main have the same +qualified tree. [Automatic main qualification is complete](https://github.com/durable-workflow/server/pull/411#issuecomment-6113537310): +all six Shared configurations, PHP/source and boundary pass on that merge. +Candidate PHP/source passes 2,683 tests/61,212 assertions; main passes +2,683 tests/61,208 assertions. Both report six existing deprecations and one +skip. Completed source resources and branches are removed. + +[#415](https://github.com/durable-workflow/server/pull/415) adds all nine +published PHP/Python/Rust signal-client and workflow-author combinations. Its +[initial correction run](https://github.com/durable-workflow/server/actions/runs/38182926674) +passes all six database configurations at `2de70bf8d029728c9cf7a8890461ae56628e34a8`, +with PHP/source and both boundaries passing. PHP reports 2,685 tests/61,266 +assertions, six existing deprecations and one skip. Independent original +[SQLite inspection](https://github.com/durable-workflow/server/pull/415#issuecomment-6115075064) +passes all 27 observations, binds all 90 HTTP phase sidecars and verifies all +133 retained original/archive member hashes. Separate +[MySQL 8.0 inspection](https://github.com/durable-workflow/server/pull/415#issuecomment-6115110451) +passes 27 observations and 135 member hashes, with actual 1 GiB database limits +and TLS checks. Both original native outputs pass all 75 execution tests. +Complete signal profiles, tests and thin SDK proof are retained; duplicate +downloads and extraction scratch are removed. Final current-main integration +qualification remains pending in #415. + +The latest integration exposes a separate frozen-PHP startup overlap: +[original SQLite failure](https://github.com/durable-workflow/server/pull/415#issuecomment-6115559242). +Five directions pass and the sixth completes its original 13-event run, but +the first worker claims `WorkflowStarted` before the successful start response; +PHP subsequently adds ten declaration fields to that same original event. +The strict full-history check fails. Original native output passes 75 tests, +but this failed profile never reaches embedded/native signal execution. +The full failed profile and wire bytes remain retained, with all 62 original +and archive-roundtrip hashes verified; duplicate downloads and scratch are removed. +The [PHP finding is handed off](https://github.com/durable-workflow/server/pull/420#issuecomment-6115559389) +without claiming a current-release reproduction or changing PHP/SDK code. +The [reviewed fixture decision](https://github.com/durable-workflow/server/pull/415#issuecomment-6115559242) +holds the original real workflow poll in the byte-preserving observer until +the matching successful start receipt. This bounds the cohort to post-start +delivery and cold replay. Full metadata/history guards remain strict; startup +overlap remains unqualified, and the failed record is not relabeled. +Focused source validation passes 61 signal models and five socket-based +transport cases; the current Node contract suite passes all 4,681 tests and +the changed PHP recorder passes syntax checking. The native Rust and SDK +adapter sources remain unchanged. Fresh runtime qualification is still required. + +The original normal worker registers its declaration before the observer starts +the run, and its first workflow claim follows the observed successful start +receipt. Two distinct normal clients send identical exact Avro payloads; three +cold author processes retain both accepted identities and full original +history through completion. The [reviewed scope](https://github.com/durable-workflow/server/issues/325#issuecomment-6113352490) +uses the instance-current-run API, preserves Python's absent client return and +retains actual server receipts. Canonical embedded direct-signal histories stay +intact; the separate manifest preserves prior tuples and resource envelopes. +Normal accepted signals require the published HTTP 202 contract. All 60 signal +reader models and all 4,482 reader tests pass, alongside the recorder, language, +shell and Docker isolation checks. Exact SDK preparation, source and package +hashes pass; those model/build results remain distinct from runtime evidence. + +Normal Rust authoring exposed two native gaps. The +[occurrence correction](https://github.com/durable-workflow/server/pull/415#issuecomment-6114363122) +retains nonempty authored condition occurrence IDs through opening, restart and +matching satisfaction, omitting absent/null/empty values as PHP does. The +[variadic correction](https://github.com/durable-workflow/server/pull/415#issuecomment-6114821113) +accepts zero or more tail values after required positional arguments, validates +each value without coercion and preserves int64/float/null distinctions. Three +database/HTTP regressions cover repeated waits, cold reopening, duplicate/stale +completions, optional occurrence types and exact variadic values/refusals. +Earlier failed records and the separate frozen SQLite patch-pressure failure +remain unchanged. Defaults, optional fixed prefixes, non-final variadics, +older-protocol occurrence routing and broader signal policies remain separate +unqualified gates. Normal published SDK query directions are the next breadth +slice; matched performance and cutover remain open. [#401](https://github.com/durable-workflow/server/pull/401) qualifies all nine fresh PHP/Python/Rust workflow-author and separately @@ -99,7 +173,9 @@ This is one authored workflow/activity scenario in the default namespace, with nine language directions; it does not establish nine capability families or performance. Separately routed parent/child SDK directions are qualified in #408. -Normal SDK timer authoring and cold replay are the next fixture slice. +Normal SDK timer authoring and cold replay are qualified in #411; normal signal +client/author directions are implemented with runtime qualification pending. +Broader signal policies and query/update clients and authors remain ahead. The observed activity-only Python polling defect is handed to [SDK Python #119](https://github.com/durable-workflow/sdk-python/issues/119). @@ -1156,7 +1232,7 @@ does not prove the remaining acceptance criteria. | Workflow lifecycle, duplicate commands, typed history, continuation, child workflows, cancellation, cleanup replay | Workflow lifecycle, migration, child-workflow and replay runners; Feature cooperative cancellation, history, migration and repeated-signal tests | Bounded authored completion, activity retry/failure, direct children/cancellation and root cooperative cleanup; broader propagation, replay and continuation semantics pending | | Timers, schedules, signals, queries, updates, search attributes, memo, sagas | Corresponding runners in `scripts/conformance/`, manifests in `static/platform-conformance/`; Sample App polyglot experiments | Bounded durable sleep, repeated signals, state queries/updates and two UTC schedule cases; broader policies, search attributes, memo and sagas pending | | Local activities, cancellation scopes, worker sessions, streams, service catalog/Nexus, bridge adapters, standalone activities, debugging and repair | `routes/api.php`, `docs/contracts/`, worker/control-plane specifications, corresponding Feature tests and Nexus runner | Pending; not omitted from parity | -| Actual PHP/Python/Rust directions and existing CLI/Waterline/Sample App | Organization [conformance runbook](https://github.com/durable-workflow/.github/blob/main/conformance/README.md) and its SDK coverage inventory; Sample App activity, child, timer, saga, update, namespace and search experiments | Published PHP SDK and embedded directions for 29 default fixtures plus a separate role profile; #401 qualifies nine PHP/Python/Rust workflow/activity directions and #408 nine parent/child directions, with cold replay and canonical embedded counterparts; unchanged CLI 2.2.0 list/describe/history; normal SDK timer workers and cold replay implemented with qualification tracked in #411; broader SDK families, CLI commands, Waterline and Sample App pending | +| Actual PHP/Python/Rust directions and existing CLI/Waterline/Sample App | Organization [conformance runbook](https://github.com/durable-workflow/.github/blob/main/conformance/README.md) and its SDK coverage inventory; Sample App activity, child, timer, saga, update, namespace and search experiments | Published PHP SDK and embedded directions for 29 default fixtures plus a separate role profile; #401 qualifies nine PHP/Python/Rust workflow/activity directions, #408 nine parent/child directions and #411 normal timer authors, with cold replay and canonical embedded counterparts; unchanged CLI 2.2.0 list/describe/history; nine signal client/author directions implemented with runtime qualification pending; broader SDK families, CLI commands, Waterline and Sample App pending | | SQL schema and stored representations | `database/migrations/`, Workflow package `src/migrations/`, model casts, `Workflow\\Serializers\\Serializer`, queue job payloads, exported history and credential digests | [Representation checkpoints](rust-server-storage-audit.md) recorded; complete mapping and executable migration tests pending | | Images, architecture, bootstrap, configuration, readiness, metrics and graceful shutdown | `Dockerfile`, `docker/`, `config/`, `docker-compose*.yml`, `k8s/helm/`, `docs/server-reference.md`, small-cluster/multi-region validation docs | Pending | | Backup and in-place upgrade | `docs/self-hosted-backup-and-restore.md`, external payload backup holds; last PHP image writes, all PHP roles stop, Rust takes over same DB sequentially | Pending on SQLite, MariaDB/MySQL, PostgreSQL | @@ -1223,8 +1299,10 @@ signals must not move replayed activity/timer/wait positions. Official Avro signal decode/encode runs in `spawn_blocking` with one permit per runtime, held until that job finishes even when its HTTP request is cancelled. Transaction serialization, full codec resource limits and CPU/deadline qualification remain -open. Initial signal argument support is positional, without default/variadic -or class-type conversion; unsupported contracts fail before durable writes. +open. Initial signal argument support was positional, without default/variadic +or class-type conversion. The current normal SDK signal slice implements +variadic tails with required positional prefixes; defaults and class-type +conversion remain unsupported and fail before durable writes. Timed/grouped waits, cancellation, full rejection audit parity, authorization and namespace coverage still require their own slices. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 43e7e3cdb..601f5beb0 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1238,8 +1238,19 @@ where "open_condition_wait" => { reject_fields( command, - &["type", "condition_key", "condition_definition_fingerprint"], + &[ + "type", + "condition_key", + "condition_definition_fingerprint", + "condition_wait_occurrence_id", + ], )?; + if command + .get("condition_wait_occurrence_id") + .is_some_and(|value| !value.is_null() && !value.is_string()) + { + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_request")); + } text(command, "condition_key")?; let fingerprint = text(command, "condition_definition_fingerprint")?; if !fingerprint.strip_prefix("sha256:").is_some_and(|hash| { @@ -1442,6 +1453,18 @@ where let direct = command["type"] == "open_signal_wait"; let mut payload = command.clone(); payload.as_object_mut().unwrap().remove("type"); + // Match PHP's optional occurrence payload: retain authored + // nonempty strings and omit null/empty legacy values. + if !direct + && payload + .get("condition_wait_occurrence_id") + .is_some_and(|value| value.is_null() || value.as_str() == Some("")) + { + payload + .as_object_mut() + .unwrap() + .remove("condition_wait_occurrence_id"); + } let wait_id = if direct { let pending = Self::query("SELECT signal_wait_id FROM workflow_signal_records WHERE workflow_run_id=$1 AND signal_name=$2 AND status='received' ORDER BY command_sequence,id LIMIT 1") .bind(&run_id).bind(text(command, "signal_name")?).fetch_optional(&mut *tx).await?; @@ -2024,22 +2047,37 @@ pub(super) fn validate_arguments( .and_then(|contracts| contracts.iter().find(|contract| contract["name"] == name)) .and_then(|contract| contract["parameters"].as_array()) .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, unsupported))?; - if parameters.len() != values.len() { + let variadic = parameters + .last() + .is_some_and(|parameter| parameter["variadic"] == true); + let fixed_count = parameters.len() - usize::from(variadic); + if parameters.iter().enumerate().any(|(index, parameter)| { + parameter["default_available"] == true + || (parameter["variadic"] == true && index + 1 != parameters.len()) + || (variadic && index < fixed_count && parameter["required"] != true) + }) { + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, unsupported)); + } + if values.len() < fixed_count || (!variadic && parameters.len() != values.len()) { return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, invalid)); } - for (parameter, value) in parameters.iter().zip(values) { - if parameter["variadic"] == true || parameter["default_available"] == true { - return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, unsupported)); - } - if matches!(value, crate::codec::Value::Null) && parameter["allows_null"] == true { - continue; + for (index, value) in values.iter().enumerate() { + let parameter = ¶meters[index.min(fixed_count)]; + if matches!(value, crate::codec::Value::Null) { + if parameter["allows_null"] + .as_bool() + .unwrap_or(parameter["type"].is_null() || parameter["type"] == "") + { + continue; + } + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, invalid)); } use crate::codec::Value as Payload; let valid = match parameter["type"].as_str() { None | Some("mixed") => true, Some("array") => matches!(value, Payload::Array(_) | Payload::Map(_)), Some("int") => matches!(value, Payload::Long(_)), - Some("float") => matches!(value, Payload::Double(_)), + Some("float") => matches!(value, Payload::Double(_) | Payload::Long(_)), Some("bool") => matches!(value, Payload::Boolean(_)), Some("string") => matches!(value, Payload::String(_)), _ => { diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index ab08c1a2f..2d24967e4 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -52,6 +52,9 @@ mod task_retry; #[path = "execution/selections.rs"] mod selections; +#[path = "execution/condition_occurrences.rs"] +mod condition_occurrences; + enum TestDatabase { Sqlite(tempfile::TempDir), Postgres { diff --git a/rust/tests/execution/condition_occurrences.rs b/rust/tests/execution/condition_occurrences.rs new file mode 100644 index 000000000..172cc3d8c --- /dev/null +++ b/rust/tests/execution/condition_occurrences.rs @@ -0,0 +1,382 @@ +use super::*; + +fn condition(occurrence: Value) -> Value { + json!({"type":"open_condition_wait","condition_key":"ready", + "condition_definition_fingerprint":format!("sha256:{}", "a".repeat(64)), + "condition_wait_occurrence_id":occurrence}) +} + +async fn register_signal_worker(app: &Router, worker: &str) { + let response = request( + app, + "POST", + "/api/worker/register", + json!({"worker_id":worker,"task_queue":"test","runtime":"rust","supported_workflow_types":["echo"], + "supported_activity_types":[], + "workflow_command_contracts":{"echo":{"signals":["payload"],"signal_contracts":[{ + "name":"payload","parameters":[{"name":"args","position":0,"type":null, + "required":false,"variadic":true,"default_available":false,"default":null,"allows_null":true}]}]}}}), + ) + .await; + assert_eq!(response.0, StatusCode::CREATED, "{}", response.1); +} + +#[tokio::test] +async fn authored_occurrences_survive_restart_and_completion_fences() { + let database = TestDatabase::new().await; + let first = database.open().await.unwrap(); + let app = router(first.clone()); + register_signal_worker(&app, "occurrence-original").await; + let started = start(&app, "occurrence-original-run").await; + let initial = poll(&app, "occurrence-original", "workflow").await; + let first_wait = json!([condition(json!("rust:condition-wait:0"))]); + assert_eq!( + finish_task(&app, &initial, first_wait.clone()).await.0, + StatusCode::OK + ); + let prefix = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!(prefix.as_array().unwrap().len(), 3); + assert_eq!( + prefix[2]["payload"]["condition_wait_occurrence_id"], + "rust:condition-wait:0" + ); + assert_eq!( + finish_task(&app, &initial, first_wait).await.1["recorded"], + false + ); + assert_eq!( + run_history(&app, &started["workflow_id"], &started["run_id"]).await, + prefix + ); + drop(app); + first.close().await; + + let second = database.open().await.unwrap(); + let app = router(second.clone()); + register_signal_worker(&app, "occurrence-resumed").await; + let path = format!( + "/api/workflows/{}/signal/payload", + started["workflow_id"].as_str().unwrap() + ); + let input = json!({"input":envelope(Payload::Array(vec![Payload::Long(9007199254740993)]))}); + let first_signal = request(&app, "POST", &path, input.clone()).await; + assert_eq!(first_signal.0, StatusCode::ACCEPTED, "{}", first_signal.1); + let resumed = poll(&app, "occurrence-resumed", "workflow").await; + assert_eq!( + resumed["history_events"][2]["payload"], + prefix[2]["payload"] + ); + let second_wait = json!([condition(json!("rust:condition-wait:1"))]); + let mut stale = completion(&resumed, second_wait.clone()); + stale["lease_owner"] = json!("occurrence-original"); + let completion_path = format!( + "/api/worker/workflow-tasks/{}/complete", + resumed["task_id"].as_str().unwrap() + ); + let before = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!( + request(&app, "POST", &completion_path, stale).await.0, + StatusCode::CONFLICT + ); + assert_eq!( + run_history(&app, &started["workflow_id"], &started["run_id"]).await, + before + ); + assert_eq!( + finish_task(&app, &resumed, second_wait.clone()).await.0, + StatusCode::OK + ); + assert_eq!( + finish_task(&app, &resumed, second_wait).await.1["recorded"], + false + ); + let middle = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!(middle.as_array().unwrap().len(), 8); + assert_eq!( + middle[6]["payload"]["condition_wait_occurrence_id"], + "rust:condition-wait:0" + ); + assert_eq!( + middle[6]["payload"]["condition_wait_id"], + prefix[2]["payload"]["condition_wait_id"] + ); + assert_eq!( + middle[7]["payload"]["condition_wait_occurrence_id"], + "rust:condition-wait:1" + ); + drop(app); + second.close().await; + + let third = database.open().await.unwrap(); + let app = router(third.clone()); + register_signal_worker(&app, "occurrence-replacement").await; + let second_signal = request(&app, "POST", &path, input).await; + assert_eq!(second_signal.0, StatusCode::ACCEPTED, "{}", second_signal.1); + let replacement = poll(&app, "occurrence-replacement", "workflow").await; + assert_eq!( + replacement["history_events"][2]["payload"], + prefix[2]["payload"] + ); + assert_eq!( + replacement["history_events"][6]["payload"], + middle[6]["payload"] + ); + assert_eq!( + replacement["history_events"][7]["payload"], + middle[7]["payload"] + ); + let result = envelope(Payload::Long(9007199254740993)); + let commands = json!([{"type":"complete_workflow","result":result}]); + assert_eq!( + finish_task(&app, &replacement, commands.clone()).await.0, + StatusCode::OK + ); + assert_eq!( + finish_task(&app, &replacement, commands).await.1["recorded"], + false + ); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert_eq!( + &history.as_array().unwrap()[..8], + middle.as_array().unwrap().as_slice() + ); + assert_eq!( + history + .as_array() + .unwrap() + .iter() + .map(|event| event["event_type"].as_str().unwrap()) + .collect::>(), + [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ] + ); + assert_eq!( + history[11]["payload"]["condition_wait_occurrence_id"], + "rust:condition-wait:1" + ); + assert_eq!( + history[11]["payload"]["condition_wait_id"], + middle[7]["payload"]["condition_wait_id"] + ); + assert_eq!(history[12]["payload"]["output"], result); + third.close().await; + database.remove().await; +} + +#[tokio::test] +async fn optional_occurrences_match_php_and_invalid_types_leave_the_attempt_usable() { + // PHP omits absent/null/empty occurrences and keeps authored strings without + // imposing the unrelated durable-ID length limit on this history metadata. + for occurrence in [ + None, + Some(Value::Null), + Some(json!("")), + Some(json!("λ".repeat(300))), + ] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "occurrence-validation", json!(["echo"]), json!([])).await; + let started = start(&app, "occurrence-validation-run").await; + let task = poll(&app, "occurrence-validation", "workflow").await; + let prefix = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + for invalid in [json!(7), json!(true), json!([]), json!({})] { + assert_eq!( + finish_task(&app, &task, json!([condition(invalid)])) + .await + .0, + StatusCode::UNPROCESSABLE_ENTITY + ); + assert_eq!( + run_history(&app, &started["workflow_id"], &started["run_id"]).await, + prefix + ); + } + let mut command = condition(occurrence.clone().unwrap_or(Value::Null)); + if occurrence.is_none() { + command + .as_object_mut() + .unwrap() + .remove("condition_wait_occurrence_id"); + } + assert_eq!( + finish_task(&app, &task, json!([command])).await.0, + StatusCode::OK + ); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let recorded = history[2]["payload"].get("condition_wait_occurrence_id"); + if occurrence + .as_ref() + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()) + { + assert_eq!(recorded, occurrence.as_ref()); + } else { + assert!(recorded.is_none()); + } + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn variadic_signal_arguments_preserve_values_and_refuse_invalid_tails() { + let tail = |position: usize, kind: Value, nullable: bool| { + json!({"name":"args","position":position,"type":kind,"required":false, + "variadic":true,"default_available":false,"default":null,"allows_null":nullable}) + }; + let prefix = json!({"name":"label","position":0,"type":"string","required":true, + "variadic":false,"default_available":false,"allows_null":false}); + let label = Payload::String("prefix λ".into()); + let map = Payload::Map(std::collections::BTreeMap::from([( + "message".into(), + Payload::String("variadic λ".into()), + )])); + for (parameters, accepted, rejected, reason) in [ + ( + json!([tail(0, Value::Null, true)]), + vec![ + vec![], + vec![Payload::Null, Payload::Long(9007199254740993), map.clone()], + ], + vec![], + "invalid_signal_arguments", + ), + ( + json!([prefix.clone(), tail(1, json!("int"), false)]), + vec![ + vec![label.clone()], + vec![ + label.clone(), + Payload::Long(9007199254740993), + Payload::Long(-9), + ], + ], + vec![ + vec![], + vec![Payload::Long(1)], + vec![label.clone(), Payload::Null], + vec![label.clone(), Payload::Boolean(true)], + vec![ + label.clone(), + Payload::Long(1), + Payload::String("wrong tail".into()), + ], + ], + "invalid_signal_arguments", + ), + ( + json!([tail(0, json!("float"), false)]), + vec![ + vec![], + vec![Payload::Long(9007199254740993), Payload::Double(1.25)], + ], + vec![ + vec![Payload::Null], + vec![Payload::Boolean(false)], + vec![Payload::String("1.25".into())], + ], + "invalid_signal_arguments", + ), + ( + json!([tail(0, Value::Null, false)]), + vec![vec![map]], + vec![vec![Payload::Null], vec![label.clone(), Payload::Null]], + "invalid_signal_arguments", + ), + ( + json!([tail(0, Value::Null, true), prefix]), + vec![], + vec![vec![], vec![Payload::Long(1)]], + "unsupported_signal_contract", + ), + ] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let registered = request( + &app, + "POST", + "/api/worker/register", + json!({"worker_id":"variadic-worker","runtime":"rust","task_queue":"test", + "supported_workflow_types":["echo"],"supported_activity_types":[], + "workflow_command_contracts":{"echo":{"signals":["payload"], + "signal_contracts":[{"name":"payload","parameters":parameters}]}}}), + ) + .await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let started = start(&app, "variadic-signal-run").await; + // Keep the original workflow task leased: accepted signals are buffered, + // while refused argument lists must add no command/history mutation. + let _task = poll(&app, "variadic-worker", "workflow").await; + let path = "/api/workflows/variadic-signal-run/signal/payload"; + for values in rejected { + let before = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let response = request( + &app, + "POST", + path, + json!({"input":envelope(Payload::Array(values))}), + ) + .await; + assert_eq!( + response.0, + StatusCode::UNPROCESSABLE_ENTITY, + "{}", + response.1 + ); + assert_eq!(response.1["reason"], reason); + assert_eq!( + run_history(&app, &started["workflow_id"], &started["run_id"]).await, + before + ); + } + for values in accepted { + let before = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let expected = Payload::Array(values); + let response = request( + &app, + "POST", + path, + json!({"input":envelope(expected.clone())}), + ) + .await; + assert_eq!(response.0, StatusCode::ACCEPTED, "{}", response.1); + let after = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let events = after.as_array().unwrap(); + assert_eq!(events.len(), before.as_array().unwrap().len() + 1); + assert_eq!( + &events[..events.len() - 1], + before.as_array().unwrap().as_slice() + ); + let event = events.last().unwrap(); + assert_eq!(event["event_type"], "SignalReceived"); + assert_eq!( + event["payload"]["workflow_command_id"], + response.1["command_id"] + ); + assert_eq!( + ValueCodec::new() + .unwrap() + .decode(event["payload"]["arguments"]["blob"].as_str().unwrap()) + .unwrap(), + expected + ); + } + runtime.close().await; + database.remove().await; + } +} diff --git a/scripts/conformance/server-parity.mjs b/scripts/conformance/server-parity.mjs index 87a0541af..634d19d5a 100644 --- a/scripts/conformance/server-parity.mjs +++ b/scripts/conformance/server-parity.mjs @@ -13,6 +13,7 @@ import {checkSdkDirectionArtifacts, checkSdkDirectionPackages} from './server-pa import {parseDatabaseResourceLimits} from './server-parity/database-resource-contract.mjs'; import {checkSdkChildDirectionArtifacts, checkSdkChildDirectionPackages} from './server-parity/sdk-child-direction-contract.mjs'; import {checkSdkTimerAuthorArtifacts, checkSdkTimerAuthorPackages} from './server-parity/sdk-timer-author-contract.mjs'; +import {checkSdkSignalDirectionArtifacts, checkSdkSignalDirectionPackages} from './server-parity/sdk-signal-direction-contract.mjs'; const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..'); const {values, positionals} = parseArgs({allowPositionals: true, options: { @@ -83,6 +84,7 @@ try { checkSdkDirectionArtifacts(fixture, record.artifacts); checkSdkChildDirectionArtifacts(fixture, record.artifacts); checkSdkTimerAuthorArtifacts(fixture, record.artifacts); + checkSdkSignalDirectionArtifacts(fixture, record.artifacts); if (fixture.patch_deployment?.legacy_marker_history) checkLegacyMarkerArtifacts(fixture, record.artifacts); const workflowId = `${values.prefix}-${fixture.id}`; const arguments_ = [resolve(root, 'scripts/conformance/server-parity/probe.php'), '--mode', values.mode, '--fixture', path, '--workflow-id', workflowId]; @@ -108,6 +110,7 @@ try { checkSdkDirectionPackages(fixture, item.observation, record.artifacts); checkSdkChildDirectionPackages(fixture, item.observation, record.artifacts); checkSdkTimerAuthorPackages(fixture, item.observation, record.artifacts); + checkSdkSignalDirectionPackages(fixture, item.observation, record.artifacts); item.outcome = 'product-fail'; item.projection = checkObservation(fixture, item.observation, workflowId); item.outcome = 'pass'; diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 3e6724f8c..6f6c22268 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -18,6 +18,7 @@ import {checkSdkDirection, checkSdkDirectionPackages} from './sdk-direction-cont import {checkDatabaseResourceLimits} from './database-resource-contract.mjs'; import {checkSdkChildDirection, checkSdkChildDirectionPackages} from './sdk-child-direction-contract.mjs'; import {checkSdkTimerAuthor, checkSdkTimerAuthorPackages} from './sdk-timer-author-contract.mjs'; +import {checkSdkSignalDirection, checkSdkSignalDirectionPackages} from './sdk-signal-direction-contract.mjs'; export function instantNanoseconds(value) { const shape = typeof value === 'string' && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.(\d{1,9}))?(?:Z|[+-]\d{2}:\d{2})$/.exec(value); @@ -40,7 +41,7 @@ export function checkObservation(fixture, observation, workflowId) { nonempty(observation.run_id, 'run identity'); equal(observation.workflow_type, fixture.workflow_type, 'registered workflow type'); equal(observation.namespace, fixture.namespace ?? 'default', 'namespace'); - equal(observation.task_queue, fixture.sdk_timer_author?.task_queue ?? fixture.sdk_child_direction?.parent_queue ?? fixture.sdk_direction?.workflow_queue ?? 'server-parity-v1', 'task queue'); + equal(observation.task_queue, fixture.sdk_signal_direction?.task_queue ?? fixture.sdk_timer_author?.task_queue ?? fixture.sdk_child_direction?.parent_queue ?? fixture.sdk_direction?.workflow_queue ?? 'server-parity-v1', 'task queue'); const cancelled = Boolean(fixture.immediate_cancellation || fixture.cooperative_cancellation); equal(observation.status, cancelled ? 'cancelled' : 'completed', 'actual durable terminal state'); equal(observation.payload_codec, 'avro', 'payload codec'); @@ -105,6 +106,7 @@ export function checkObservation(fixture, observation, workflowId) { const sdkDirection = checkSdkDirection(fixture, observation, workflowId); const sdkChildDirection = checkSdkChildDirection(fixture, observation, workflowId); const sdkTimerAuthor = checkSdkTimerAuthor(fixture, observation, workflowId); + const sdkSignalDirection = checkSdkSignalDirection(fixture, observation, workflowId); const cooperative = fixture.cooperative_cancellation ? checkCooperativeCancellation(fixture, observation, identities, projectedEvents, instantNanoseconds) : null; if (fixture.child_cancellation) { @@ -584,6 +586,7 @@ export function checkObservation(fixture, observation, workflowId) { ...(sdkDirection ? {sdk_direction: sdkDirection} : {}), ...(sdkChildDirection ? {sdk_child_direction: sdkChildDirection} : {}), ...(sdkTimerAuthor ? {sdk_timer_author: sdkTimerAuthor} : {}), + ...(sdkSignalDirection ? {sdk_signal_direction: sdkSignalDirection} : {}), }; } @@ -618,6 +621,7 @@ export function compareRecords(records, expectedFixtureHashes, expectedFixtures, checkSdkDirectionPackages(item.fixture, item.observation, record.artifacts); checkSdkChildDirectionPackages(item.fixture, item.observation, record.artifacts); checkSdkTimerAuthorPackages(item.fixture, item.observation, record.artifacts); + checkSdkSignalDirectionPackages(item.fixture, item.observation, record.artifacts); if (item.fixture.patch_deployment?.legacy_marker_history) { checkLegacyMarkerArtifacts(item.fixture, record.artifacts); checkLegacyMarkerPackageObservation(item.fixture, item.observation, record.artifacts); diff --git a/scripts/conformance/server-parity/embedded-types.php b/scripts/conformance/server-parity/embedded-types.php index 0bb9927ee..676dc6e4a 100644 --- a/scripts/conformance/server-parity/embedded-types.php +++ b/scripts/conformance/server-parity/embedded-types.php @@ -326,6 +326,21 @@ public function handle(array $value, int $target): array } } +#[Type('parity.v1.sdk_signals')] +#[Signal('payload', [['name' => 'value', 'type' => 'array']])] +final class SdkSignalsWorkflow extends Workflow +{ + public function handle(array $value, int $target): array + { + $received = null; + for ($index = 0; $index < $target; $index++) { + $received = signal('payload'); + } + + return $received; + } +} + #[Type('parity.v1.queries')] #[Signal('payload', [['name' => 'value', 'type' => 'array']])] final class QueriesWorkflow extends Workflow diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index b685d607c..1dc9f0cd6 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -21,6 +21,7 @@ use ServerParity\TimerWorkflow; use ServerParity\SdkTimerWorkflow; use ServerParity\SignalsWorkflow; +use ServerParity\SdkSignalsWorkflow; use ServerParity\QueriesWorkflow; use ServerParity\UpdatesWorkflow; use ServerParity\TwoChildrenWorkflow; @@ -51,7 +52,7 @@ $fixture = json_decode(file_get_contents($options['fixture']), true, flags: JSON_THROW_ON_ERROR); $workflowId = $options['workflow-id']; $namespace = 'default'; -$queue = $fixture['sdk_timer_author']['task_queue'] ?? $fixture['sdk_child_direction']['parent_queue'] ?? $fixture['sdk_direction']['workflow_queue'] ?? 'server-parity-v1'; +$queue = $fixture['sdk_signal_direction']['task_queue'] ?? $fixture['sdk_timer_author']['task_queue'] ?? $fixture['sdk_child_direction']['parent_queue'] ?? $fixture['sdk_direction']['workflow_queue'] ?? 'server-parity-v1'; $sdkAutoload = getenv('DW_PARITY_SDK_AUTOLOAD') ?: __DIR__.'/vendor/autoload.php'; if (! is_file($sdkAutoload)) { fwrite(STDERR, "Install the locked parity SDK adapter first.\n"); @@ -547,6 +548,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa 'parity.v1.one_timer' => TimerWorkflow::class, 'parity.v1.sdk_timer' => SdkTimerWorkflow::class, 'parity.v1.signals' => SignalsWorkflow::class, + 'parity.v1.sdk_signals' => SdkSignalsWorkflow::class, 'parity.v1.queries' => QueriesWorkflow::class, 'parity.v1.updates' => UpdatesWorkflow::class, 'parity.v1.two_children' => TwoChildrenWorkflow::class, @@ -576,6 +578,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if (isset($fixture['sdk_timer_author'])) { $class = SdkTimerWorkflow::class; } + if (isset($fixture['sdk_signal_direction'])) { + $class = SdkSignalsWorkflow::class; + } if (isset($fixture['retry_policy'])) { $class = ActivityRetryWorkflow::class; } @@ -786,13 +791,13 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/child-cancellation-probe.php'; require __DIR__.'/cancellation-probe.php'; require __DIR__.'/cooperative-probe.php'; - if (isset($fixture['sdk_direction']) || isset($fixture['sdk_child_direction']) || isset($fixture['sdk_timer_author'])) { + if (isset($fixture['sdk_direction']) || isset($fixture['sdk_child_direction']) || isset($fixture['sdk_timer_author']) || isset($fixture['sdk_signal_direction'])) { require __DIR__.'/sdk-direction-probe.php'; if ($mode === 'http') { $observation = sdkDirectionHttpObservation($fixture, $options); } elseif ($mode === 'embedded') { $observation = embeddedObservation($fixture, $workflowId, $namespace, $queue, $options['application-root']); - $observation[isset($fixture['sdk_timer_author']) ? 'sdk_timer_author' : (isset($fixture['sdk_child_direction']) ? 'sdk_child_direction' : 'sdk_direction')] + $observation[isset($fixture['sdk_signal_direction']) ? 'sdk_signal_direction' : (isset($fixture['sdk_timer_author']) ? 'sdk_timer_author' : (isset($fixture['sdk_child_direction']) ? 'sdk_child_direction' : 'sdk_direction'))] = ['applicable' => false, 'reason' => 'embedded_executes_php_author_definitions']; } else { throw new InvalidArgumentException('Mode must be http or embedded.'); @@ -857,7 +862,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && ! isset($observation['workflow_package'])) { $observation['workflow_package'] = ltrim(InstalledVersions::getPrettyVersion('durable-workflow/workflow'), 'v'); } - if (isset($fixture['sdk_direction']) || isset($fixture['sdk_child_direction']) || isset($fixture['sdk_timer_author'])) { + if (isset($fixture['sdk_direction']) || isset($fixture['sdk_child_direction']) || isset($fixture['sdk_timer_author']) || isset($fixture['sdk_signal_direction'])) { sdkDirectionPackageObservation($observation, $sdkAutoload, $mode === 'embedded' ? $options['application-root'] : null); } if ($mode === 'embedded' && isset($fixture['patch_deployment']['embedded_clock_probe']) && ! isset($observation['workflow_source'])) { diff --git a/scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs b/scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs index 0b6768492..24f55ef7b 100644 --- a/scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs +++ b/scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs @@ -3,13 +3,18 @@ import http from 'node:http'; import {appendFileSync} from 'node:fs'; import {gunzipSync, inflateSync, brotliDecompressSync} from 'node:zlib'; -const [target, journal, observationMode] = process.argv.slice(2); +const [target, journal, observationMode, gateMode, gateWorkflowId, ...extra] = process.argv.slice(2); if (observationMode !== undefined && observationMode !== '--forwarding-state') throw new Error('Unknown observation mode'); +if (gateMode !== undefined && (gateMode !== '--workflow-poll-after-start' || !gateWorkflowId || extra.length + || observationMode !== '--forwarding-state')) throw new Error('Invalid workflow start gate'); const upstream = new URL(target); if (upstream.protocol !== 'http:') throw new Error('Isolated fixture target must use HTTP'); +let startObserved = gateMode === undefined; +const waitingPolls = new Set(); const server = http.createServer((request, response) => { const chunks = []; let forwarded; + let forwardRequest; let requestBodyComplete = false; let recorded = false; const record = fields => { @@ -27,6 +32,7 @@ const server = http.createServer((request, response) => { response.destroy(error); }; response.on('close', () => { + waitingPolls.delete(forwardRequest); if (!recorded && !response.writableFinished) { record({client_cancelled: true}); forwarded?.destroy(); @@ -37,28 +43,48 @@ const server = http.createServer((request, response) => { requestBodyComplete = true; if (recorded) return; const body = Buffer.concat(chunks); - forwarded = http.request(new URL(request.url, upstream), { - method: request.method, - headers: {...request.headers, host: upstream.host}, - }, incoming => { - const received = []; - incoming.on('data', chunk => received.push(chunk)); - incoming.on('end', () => { - if (recorded) return; - const bytes = Buffer.concat(received); - const encoding = incoming.headers['content-encoding'] ?? 'identity'; - const decoders = {identity: value => value, gzip: gunzipSync, deflate: inflateSync, br: brotliDecompressSync}; - if (!decoders[encoding]) throw new Error(`Unsupported observed content encoding: ${encoding}`); - record({status: incoming.statusCode, response_encoding: encoding, - response_retry_after: incoming.headers['retry-after'] ?? null, - response_bytes_base64: bytes.toString('base64'), response_body: decoders[encoding](bytes).toString('utf8')}); - response.writeHead(incoming.statusCode, incoming.headers); - response.end(bytes); + forwardRequest = () => { + if (recorded) return; + forwarded = http.request(new URL(request.url, upstream), { + method: request.method, + headers: {...request.headers, host: upstream.host}, + }, incoming => { + const received = []; + incoming.on('data', chunk => received.push(chunk)); + incoming.on('end', () => { + if (recorded) return; + const bytes = Buffer.concat(received); + const encoding = incoming.headers['content-encoding'] ?? 'identity'; + const decoders = {identity: value => value, gzip: gunzipSync, deflate: inflateSync, br: brotliDecompressSync}; + if (!decoders[encoding]) throw new Error(`Unsupported observed content encoding: ${encoding}`); + const decoded = decoders[encoding](bytes).toString('utf8'); + record({status: incoming.statusCode, response_encoding: encoding, + response_retry_after: incoming.headers['retry-after'] ?? null, + response_bytes_base64: bytes.toString('base64'), response_body: decoded}); + response.writeHead(incoming.statusCode, incoming.headers); + response.end(bytes); + if (!startObserved && request.method === 'POST' && new URL(request.url, upstream).pathname === '/api/workflows' + && incoming.statusCode === 201) { + let started, receipt; + try { started = JSON.parse(body.toString('utf8')); receipt = JSON.parse(decoded); } catch { return; } + if (started.workflow_id === gateWorkflowId && typeof receipt.run_id === 'string' && receipt.run_id.length > 0) { + startObserved = true; + const pending = [...waitingPolls]; + waitingPolls.clear(); + for (const forward of pending) forward(); + } + } + }); + incoming.on('error', failed); }); - incoming.on('error', failed); - }); - forwarded.on('error', failed); - forwarded.end(body); + forwarded.on('error', failed); + forwarded.end(body); + }; + // Keep the real worker request intact until this fixture's accepted start + // is fully observed. Other cohorts retain their original forwarding behavior. + if (!startObserved && request.method === 'POST' + && new URL(request.url, upstream).pathname === '/api/worker/workflow-tasks/poll') waitingPolls.add(forwardRequest); + else forwardRequest(); }); }); server.listen(0, '127.0.0.1', () => console.log(`http://127.0.0.1:${server.address().port}`)); diff --git a/scripts/conformance/server-parity/sdk-direction-cleanup.sh b/scripts/conformance/server-parity/sdk-direction-cleanup.sh index 785f949fd..29dbe5b69 100644 --- a/scripts/conformance/server-parity/sdk-direction-cleanup.sh +++ b/scripts/conformance/server-parity/sdk-direction-cleanup.sh @@ -8,7 +8,8 @@ case "${2:-activity}" in activity) ;; child) tag=sdk-child-directions; cohort=published-sdk-child-directions ;; timer) tag=sdk-timers; cohort=published-sdk-timers ;; - *) echo 'Choose activity, child or timer SDK authors' >&2; exit 2 ;; + signal) tag=sdk-signals; cohort=published-sdk-signals ;; + *) echo 'Choose activity, child, timer or signal SDK directions' >&2; exit 2 ;; esac test -n "${RESOURCE_SCOPE:?Set the owning isolated network/resource scope}" profile="tests/Fixtures/ServerParityProfiles/$cohort" diff --git a/scripts/conformance/server-parity/sdk-direction-probe.php b/scripts/conformance/server-parity/sdk-direction-probe.php index e21a1b344..fd1318539 100644 --- a/scripts/conformance/server-parity/sdk-direction-probe.php +++ b/scripts/conformance/server-parity/sdk-direction-probe.php @@ -52,7 +52,11 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array } $name = preg_replace('/[^A-Za-z0-9_.-]/', '_', $options['workflow-id']); $journal = tempnam(sys_get_temp_dir(), 'parity-direction-'); - $proxy = proc_open(['node', __DIR__.'/published-sdk-patch/observe-http.mjs', $options['url'], $journal, '--forwarding-state'], + $proxyArguments = ['node', __DIR__.'/published-sdk-patch/observe-http.mjs', $options['url'], $journal, '--forwarding-state']; + if (isset($fixture['sdk_signal_direction'])) { + array_push($proxyArguments, '--workflow-poll-after-start', $options['workflow-id']); + } + $proxy = proc_open($proxyArguments, [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $proxyPipes); if (! is_resource($proxy)) { unlink($journal); @@ -63,19 +67,23 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array $runId = ''; $childDirection = isset($fixture['sdk_child_direction']); $timerAuthor = isset($fixture['sdk_timer_author']); - $directionKey = $timerAuthor ? 'sdk_timer_author' : ($childDirection ? 'sdk_child_direction' : 'sdk_direction'); + $signalDirection = isset($fixture['sdk_signal_direction']); + $directionKey = $signalDirection ? 'sdk_signal_direction' : ($timerAuthor ? 'sdk_timer_author' : ($childDirection ? 'sdk_child_direction' : 'sdk_direction')); $spec = $fixture[$directionKey]; $childWorkflowId = $childRunId = ''; $client = new Client($options['url'], namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null); $deadline = microtime(true) + 35; + $observerStart = null; try { $url = trim(fgets($proxyPipes[1])); if (! preg_match('#^http://127\\.0\\.0\\.1:[0-9]+$#', $url)) { throw new RuntimeException('SDK observer did not bind its local port.'); } + $startClient = $signalDirection ? new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null) : null; foreach ($spec['phase_order'] as $phase) { - $language = $timerAuthor ? $spec['language'] : ($childDirection ? $spec[$phase === 'child' ? 'child' : 'parent'] - : $spec[$phase === 'activity' ? 'activity' : 'workflow']); + $language = $signalDirection ? $spec[in_array($phase, ['signal-1', 'signal-2'], true) ? 'client' : 'workflow'] + : ($timerAuthor ? $spec['language'] : ($childDirection ? $spec[$phase === 'child' ? 'child' : 'parent'] + : $spec[$phase === 'activity' ? 'activity' : 'workflow'])); $actorWorkflowId = $phase === 'child' ? $childWorkflowId : $options['workflow-id']; $actorRunId = $phase === 'child' ? $childRunId : $runId; $arguments = match ($language) { @@ -92,20 +100,72 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array throw new RuntimeException('Could not start the published SDK phase.'); } fclose($pipes[0]); + $observedStatus = null; + $observerError = null; try { - $stdout = stream_get_contents($pipes[1]); - $stderr = stream_get_contents($pipes[2]); + if ($signalDirection && $phase === 'original') { + // Admit signals from the real worker's captured declaration. + // The observer starts only after its successful registration. + stream_set_blocking($pipes[1], false); + stream_set_blocking($pipes[2], false); + $stdout = $stderr = ''; + while (true) { + $stdout .= stream_get_contents($pipes[1]); + $stderr .= stream_get_contents($pipes[2]); + $processStatus = proc_get_status($process); + if (! $processStatus['running']) { + $observedStatus = $processStatus['exitcode']; + $stdout .= stream_get_contents($pipes[1]); + $stderr .= stream_get_contents($pipes[2]); + break; + } + if (microtime(true) >= $deadline) { + throw new RuntimeException('Signal original worker exceeded the remaining fixture budget.'); + } + if ($observerStart === null) { + foreach (array_slice(file($journal, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES), $offset) as $line) { + $wire = json_decode($line, true); + if (! is_array($wire) || $wire['method'] !== 'POST' || $wire['path'] !== '/api/worker/register' || $wire['status'] !== 201) { + continue; + } + $registration = json_decode($wire['request_body'], true, flags: JSON_THROW_ON_ERROR); + $receipt = json_decode($wire['response_body'], true, flags: JSON_THROW_ON_ERROR); + if (($registration['worker_id'] ?? null) !== $actorWorkflowId.':original' || ($receipt['registered'] ?? null) !== true) { + continue; + } + $runId = $startClient->startWorkflow($fixture['workflow_type'], $actorWorkflowId, + $spec['task_queue'], [$fixture['input'], $fixture['signal_count']])->selectedRunId; + $observerStart = ['role' => 'observer', 'language' => 'php', 'pid' => getmypid(), + 'sdk_version' => ltrim(InstalledVersions::getPrettyVersion('durable-workflow/sdk'), 'v'), + 'run_id' => $runId]; + break; + } + } + usleep(10000); + } + } else { + $stdout = stream_get_contents($pipes[1]); + $stderr = stream_get_contents($pipes[2]); + } + } catch (Throwable $error) { + if (! $signalDirection) { + throw $error; + } + $observerError = $error::class.': '.$error->getMessage(); + proc_terminate($process); } finally { fclose($pipes[1]); fclose($pipes[2]); } - $status = proc_close($process); + $closedStatus = proc_close($process); + $status = $observedStatus ?? $closedStatus; $lines = array_slice(file($journal, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES), $offset); - file_put_contents($evidence.'/'.($timerAuthor ? 'sdk-timer-author-' : ($childDirection ? 'sdk-child-direction-' : 'sdk-direction-')).$name.'-'.$phase.'.json', json_encode([ + file_put_contents($evidence.'/'.($signalDirection ? 'sdk-signal-direction-' : ($timerAuthor ? 'sdk-timer-author-' : ($childDirection ? 'sdk-child-direction-' : 'sdk-direction-'))).$name.'-'.$phase.'.json', json_encode([ 'language' => $language, 'phase' => $phase, 'stdout' => $stdout, 'stderr' => $stderr, 'exit_status' => $status, 'transport_jsonl' => implode("\n", $lines)."\n", + ...($signalDirection && $phase === 'original' ? ['observer_start' => $observerStart, 'observer_error' => $observerError] : []), ], JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE)."\n"); - $requests = array_map(static function (string $line) use ($childDirection, $timerAuthor): array { + $requests = array_map(static function (string $line) use ($childDirection, $timerAuthor, $signalDirection): array { $item = json_decode($line, true, flags: JSON_THROW_ON_ERROR); foreach (['request', 'response'] as $side) { $body = $item[$side.'_body']; @@ -116,13 +176,17 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array } // Exact request/response bytes remain in the retained sidecar. unset($item['request_body'], $item['response_body'], $item['response_bytes_base64']); - $item['response'] = sdkDirectionResponseSummary($item['response'], $childDirection || $timerAuthor); + $signalReceipt = $signalDirection && $item['method'] === 'POST' + && preg_match('#^/api/workflows/[^/]+/signal/payload$#', $item['path']); + if (! $signalReceipt) { + $item['response'] = sdkDirectionResponseSummary($item['response'], $childDirection || $timerAuthor || $signalDirection); + } return $item; }, $lines); $result = $status === 0 ? json_decode($stdout, true, flags: JSON_THROW_ON_ERROR) : ['phase' => $phase, 'language' => $language, 'worker_finished' => false, 'worker_error' => trim($stderr)]; - if ((! $childDirection && ! $timerAuthor) || $phase === 'original') { + if ((! $childDirection && ! $timerAuthor && ! $signalDirection) || $phase === 'original') { $runId = $result['run_id'] ?? $runId; } if ($runId === '') { @@ -150,7 +214,8 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array } } $phases[] = $result + ['requests' => $requests, 'snapshot' => $snapshot, - ...($childDirection ? ['child_snapshot' => $childSnapshot] : [])]; + ...($childDirection ? ['child_snapshot' => $childSnapshot] : []), + ...($observerError !== null ? ['observer_error' => $observerError] : [])]; if ($status !== 0) { break; // Preserve the actual open run and let the reader classify the product failure. } @@ -158,11 +223,31 @@ function sdkDirectionHttpObservation(array $fixture, array $options): array $observation = $phases[array_key_last($phases)]['snapshot']; $identityPath = getenv('DW_PARITY_DIRECTION_IDENTITY') ?: '/sdk-directions/identity.json'; $observation[$directionKey] = ['applicable' => true, 'phases' => $phases, - 'packages' => json_decode(file_get_contents($identityPath), true, flags: JSON_THROW_ON_ERROR)]; + 'packages' => json_decode(file_get_contents($identityPath), true, flags: JSON_THROW_ON_ERROR), + ...($signalDirection ? ['observer_start' => $observerStart] : [])]; if ($childDirection) { $observation['children'] = [$phases[array_key_last($phases)]['child_snapshot']]; } - if ($childDirection || $timerAuthor) { + if ($signalDirection) { + $observation['signal_deliveries'] = []; + foreach ($phases as $index => &$phase) { + if (! in_array($phase['phase'], ['signal-1', 'signal-2'], true)) { + continue; + } + foreach ($phase['requests'] as &$request) { + if ($request['method'] === 'POST' && $request['path'] === '/api/workflows/'.$options['workflow-id'].'/signal/payload') { + $request['decoded_input'] = $client->payloadCodec()->decodeEnvelope($request['request']['input']); + $request['typed_decoded_input'] = typedValue($request['decoded_input']); + $observation['signal_deliveries'][] = ['before' => $phases[$index - 1]['snapshot']['execution'], + 'response' => $request['response']]; + } + } + unset($request); + } + unset($phase); + $observation[$directionKey]['phases'] = $phases; + } + if ($childDirection || $timerAuthor || $signalDirection) { $observation['workflow_polls'] = $observation['workflow_completions'] = []; foreach ($phases as $phase) { foreach ($phase['requests'] as $request) { diff --git a/scripts/conformance/server-parity/sdk-direction-profile.sh b/scripts/conformance/server-parity/sdk-direction-profile.sh index e68fbebec..2c3a2df64 100644 --- a/scripts/conformance/server-parity/sdk-direction-profile.sh +++ b/scripts/conformance/server-parity/sdk-direction-profile.sh @@ -26,7 +26,14 @@ case "${3:-activity}" in fixture_pattern=timer-sdk-*.json reader_tests=(tests/Unit/ServerParitySdkTimerAuthorContractTest.mjs) ;; - *) echo 'Choose activity, child or timer SDK authors' >&2; exit 2 ;; + signal) + tag=sdk-signals + prefix=sdk-signals-v1 + cohort=published-sdk-signals + fixture_pattern=signal-sdk-*.json + reader_tests=(tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs) + ;; + *) echo 'Choose activity, child, timer or signal SDK directions' >&2; exit 2 ;; esac profile="tests/Fixtures/ServerParityProfiles/$cohort" manifest="tests/Fixtures/ServerParityProfiles/$cohort/artifacts.json" diff --git a/scripts/conformance/server-parity/sdk-directions/SIGNALS.md b/scripts/conformance/server-parity/sdk-directions/SIGNALS.md new file mode 100644 index 000000000..52c9cd7e7 --- /dev/null +++ b/scripts/conformance/server-parity/sdk-directions/SIGNALS.md @@ -0,0 +1,96 @@ +# Published SDK signal clients and cold authors + +The reviewed additive definitions cover all nine PHP/Python/Rust signal-client +and workflow-author combinations. They reuse the timer/child published tuple +without changing prior fixtures or envelopes. Normal published client/worker +adapters, canonical embedded execution, recorder guards and all six database +profiles are implemented. The initial native correction passes all six runtime +configurations; final current-main integration qualification remains pending +in [#415](https://github.com/durable-workflow/server/pull/415). + +The original normal worker registers its signal declaration before the PHP +observer starts the original run. The original worker's first workflow poll +is forwarded only after the matching real successful start receipt. +The byte-preserving observer holds that original request without changing its +bytes or synthesizing a result. Three distinct author processes open the +first condition wait, cold-replay the first signal into a second wait, then +cold-replay both accepted signals and complete that same run. Two separate +published client processes send identical Unicode/int64/boolean/null values. + +Use the normal instance-current-run signal API. Frozen Python returns no +receipt object and has no run selector; inspect the actual HTTP receipt and +durable history against the original run. Do not invent a Python return value +or change an SDK. The observer start is identified separately from the SDK +worker, and its actual request must follow successful original registration. +Normal accepted signal submissions require HTTP 202. Registration and start +retain HTTP 201; task claims and fenced completions retain HTTP 200. The +[first runtime discrepancy](https://github.com/durable-workflow/server/pull/415#issuecomment-6114095382) +was the new reader's incorrect HTTP 201 signal expectation, corrected to the +published PHP and native contract while preserving the failed record. + +Require distinct accepted signal and command IDs, ordered acceptance/application +and cursor advances, original wait identities, full unchanged history prefixes, +including the frozen Rust author's `rust:condition-wait:0` and +`rust:condition-wait:1` occurrence identities in opened and satisfied waits, +original queue/run/namespace and lease authority, real registration/claim/fenced +completion receipts, exact typed Avro results and the stored result frame. +Retain complete phase wire sidecars. Canonical embedded execution uses direct +signal waits and explicitly marks SDK language transport inapplicable; preserve +its different full event inventory before comparing common signal semantics. +The common projection contains delivery and typed-completion semantics only. +Three distinct cold SDK author processes are HTTP-specific proof; embedded +execution explicitly marks those SDK transport roles inapplicable. + +Qualification requires all six database configurations, exact installed package +and runner identities, inspected resource limits and independent runtime/wire +inspection. Reuse one shared SDK preparation and the native binary. The owning +job must clean the isolated profile through its always-run path. + +Local source checks pass: 60 signal reader models, all 4,482 reader models, +PHP recorder regressions (5 tests/24 assertions), PHP/JavaScript syntax, +Python parsing, changed-file Rust formatting, shell syntax and the Docker isolation policy +with 72 regression tests. SDK preparation passes with exact source and package +hashes. The initial [correction run](https://github.com/durable-workflow/server/actions/runs/38182926674) +passes all six database configurations, PHP/source and both boundaries. +Independent original SQLite inspection passes all 27 observations, binds all +90 HTTP sidecars and verifies 133 original/archive member hashes. Separate +MySQL 8.0 inspection passes 27 observations and 135 member hashes with actual +database limits and TLS checks. Both native outputs pass all 75 execution tests. +Complete signal profiles and thin SDK proof are retained; duplicate downloads +and scratch are removed. Final current-main strict qualification remains ahead. + +The new start-order correction passes 61 signal models plus five actual-socket +transport tests, including unrelated/rejected starts, unchanged request/response +bytes and cancellation of a fully received held poll. The complete current +Node contract suite passes 4,681 tests, and the changed PHP recorder passes +syntax checking. Native Rust and installed SDK adapter sources are unchanged; +these source checks do not qualify the new runtime ordering. + +Normal Rust authoring exposed native rejection of authored +`condition_wait_occurrence_id` and variadic signal arguments. The +[occurrence decision](https://github.com/durable-workflow/server/pull/415#issuecomment-6114363122) +retains nonempty string metadata through original wait satisfaction and omits +null/empty values as PHP does. The +[variadic decision](https://github.com/durable-workflow/server/pull/415#issuecomment-6114821113) +accepts zero or more tail values after required positional prefixes, validates +each value without coercion and respects nullability. Three HTTP/database +regressions pass for cold reopening/fences, optional metadata/types and exact +variadic values/refusals. Defaults, optional fixed prefixes and non-final +variadics remain unsupported; older-protocol occurrence routing is unqualified. +Earlier failed records and the separate frozen SQLite patch baseline pressure +failure remain unchanged with strict assertions intact. + +The latest original SQLite run also catches a frozen-PHP startup overlap: +[retained failure and contract decision](https://github.com/durable-workflow/server/pull/415#issuecomment-6115559242). +Five directions pass; the sixth completes its 13-event original run, but its +first claimed `WorkflowStarted` lacks ten declaration fields present afterward. +The raw poll response precedes the successful start response. All 62 complete +failed-profile member/archive hashes pass; original native output separately +passes 75 tests, while embedded/native SIG profiles are not reached. The full +failure is retained and the [PHP finding handed off](https://github.com/durable-workflow/server/pull/420#issuecomment-6115559389). +The new first-poll ordering excludes this startup interleaving explicitly and +preserves all full-history checks. It requires fresh six-profile qualification; +earlier all-six passes do not qualify this changed fixture. + +These definitions do not qualify run-targeted calls, continuation/current-run +races, all signal policies, CPU-load fairness, performance or cutover. diff --git a/scripts/conformance/server-parity/sdk-directions/php.php b/scripts/conformance/server-parity/sdk-directions/php.php index efc56fe7a..2954ac642 100644 --- a/scripts/conformance/server-parity/sdk-directions/php.php +++ b/scripts/conformance/server-parity/sdk-directions/php.php @@ -18,7 +18,8 @@ $fixture = json_decode(file_get_contents($fixturePath), true, flags: JSON_THROW_ON_ERROR); $childDirection = isset($fixture['sdk_child_direction']); $timerAuthor = isset($fixture['sdk_timer_author']); -if (! in_array($phase, $timerAuthor ? ['original', 'replacement'] : ($childDirection ? ['original', 'child', 'replacement'] : ['original', 'activity', 'replacement']), true)) { +$signalDirection = isset($fixture['sdk_signal_direction']); +if (! in_array($phase, $signalDirection ? ['original', 'signal-1', 'resumed', 'signal-2', 'replacement'] : ($timerAuthor ? ['original', 'replacement'] : ($childDirection ? ['original', 'child', 'replacement'] : ['original', 'activity', 'replacement'])), true)) { throw new InvalidArgumentException('Unknown SDK direction phase'); } $deadline = microtime(true) + (float) $remaining; @@ -41,13 +42,22 @@ }); $client = new Client($url, namespace: 'default', token: getenv('DW_PARITY_TOKEN') ?: null, transport: new Psr18Transport(client: new GuzzleClient(['http_errors' => false, 'handler' => $stack]))); -if ($phase === 'original') { +if ($signalDirection && in_array($phase, ['signal-1', 'signal-2'], true)) { + $result = $client->signalWorkflow($workflowId, 'payload', [$fixture['signal_value']]); + echo json_encode(['pid' => getmypid(), 'language' => 'php', + 'sdk_version' => ltrim(InstalledVersions::getPrettyVersion('durable-workflow/sdk'), 'v'), + 'phase' => $phase, 'workflow_id' => $workflowId, 'run_id' => $runId, + 'signal_sent' => true, 'sender_finished' => true, 'client_result' => $result], + JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE)."\n"; + exit(0); +} +if ($phase === 'original' && ! $signalDirection) { $runId = $client->startWorkflow($fixture['workflow_type'], $workflowId, $timerAuthor ? 'server-parity-sdk-timer' : ($childDirection ? 'server-parity-sdk-parent' : 'server-parity-sdk-workflow'), [$fixture['input']])->selectedRunId; } -$queue = $timerAuthor ? 'server-parity-sdk-timer' : ($childDirection +$queue = $signalDirection ? 'server-parity-sdk-signals' : ($timerAuthor ? 'server-parity-sdk-timer' : ($childDirection ? ($phase === 'child' ? 'server-parity-sdk-child' : 'server-parity-sdk-parent') - : ($phase === 'activity' ? 'server-parity-sdk-activity' : 'server-parity-sdk-workflow')); + : ($phase === 'activity' ? 'server-parity-sdk-activity' : 'server-parity-sdk-workflow'))); $workerId = $workflowId.':'.$phase; $worker = new Worker($client, $queue, workerId: $workerId, clock: static function () use ($deadline): float { @@ -58,7 +68,17 @@ return microtime(true); }); $workflowInvocations = $activityInvocations = 0; -if ($timerAuthor) { +if ($signalDirection) { + $worker->registerWorkflow('parity.v1.sdk_signals', static function (WorkflowContext $ctx, mixed $value, int $target) use (&$workflowInvocations, &$runId): mixed { + ++$workflowInvocations; + $runId = $ctx->runId; + for ($index = 0; $index < $target; ++$index) { + $ctx->waitCondition(fn (): bool => count($ctx->signals('payload')) > $index, 'payload:'.$index); + } + + return $ctx->signals('payload')[$target - 1][0]; + })->declareSignal('parity.v1.sdk_signals', 'payload', static fn (array $value) => null); +} elseif ($timerAuthor) { $worker->registerWorkflow('parity.v1.sdk_timer', static function (WorkflowContext $ctx, mixed $value) use (&$workflowInvocations): mixed { ++$workflowInvocations; $ctx->sleep($value['sleep_ms'] / 1000); diff --git a/scripts/conformance/server-parity/sdk-directions/python.py b/scripts/conformance/server-parity/sdk-directions/python.py index f38b3be37..7116e226d 100644 --- a/scripts/conformance/server-parity/sdk-directions/python.py +++ b/scripts/conformance/server-parity/sdk-directions/python.py @@ -56,41 +56,80 @@ def run(self, ctx, value): return value +@workflow.defn(name="parity.v1.sdk_signals") +class Signals: + def __init__(self): + self.payloads = [] + + @workflow.signal("payload") + def receive(self, value): + self.payloads.append(value) + + def run(self, ctx, value, target): + global WORKFLOW_INVOCATIONS + WORKFLOW_INVOCATIONS += 1 + for index in range(target): + yield ctx.wait_condition(lambda index=index: len(self.payloads) > index, + key="payload:" + str(index)) + return self.payloads[-1] + + async def main(): phase, url, workflow_id, run_id, fixture_path, remaining = sys.argv[1:] with open(fixture_path, encoding="utf-8") as source: fixture = json.load(source) child_direction = "sdk_child_direction" in fixture timer_author = "sdk_timer_author" in fixture - if phase not in (("original", "replacement") if timer_author else + signal_direction = "sdk_signal_direction" in fixture + if phase not in (("original", "signal-1", "resumed", "signal-2", "replacement") if signal_direction else + (("original", "replacement") if timer_author else (("original", "child", "replacement") if child_direction else - ("original", "activity", "replacement"))): + ("original", "activity", "replacement")))): raise ValueError("Unknown SDK direction phase") - queue = "server-parity-sdk-timer" if timer_author else (("server-parity-sdk-child" if phase == "child" else "server-parity-sdk-parent") + queue = "server-parity-sdk-signals" if signal_direction else ("server-parity-sdk-timer" if timer_author else (("server-parity-sdk-child" if phase == "child" else "server-parity-sdk-parent") if child_direction else - ("server-parity-sdk-activity" if phase == "activity" else "server-parity-sdk-workflow")) + ("server-parity-sdk-activity" if phase == "activity" else "server-parity-sdk-workflow"))) worker_id = workflow_id + ":" + phase async with asyncio.timeout(float(remaining)): async with Client(url, token=os.environ.get("DW_PARITY_TOKEN"), namespace="default") as client: - if phase == "original": + if signal_direction and phase in ("signal-1", "signal-2"): + result = await client.signal_workflow(workflow_id, "payload", args=[fixture["signal_value"]]) + print(json.dumps({"pid": os.getpid(), "language": "python", + "sdk_version": importlib.metadata.version("durable-workflow"), + "phase": phase, "workflow_id": workflow_id, "run_id": run_id, + "signal_sent": True, "sender_finished": True, "client_result": result})) + return + if phase == "original" and not signal_direction: handle = await client.start_workflow(workflow_type=fixture["workflow_type"], task_queue=queue, workflow_id=workflow_id, input=[fixture["input"]]) run_id = handle.run_id worker = Worker(client, task_queue=queue, worker_id=worker_id, poll_timeout=1, - workflows=[Timer] if timer_author else (([Child] if phase == "child" else [Parent]) if child_direction - else ([] if phase == "activity" else [Author])), + workflows=[Signals] if signal_direction else ([Timer] if timer_author else (([Child] if phase == "child" else [Parent]) if child_direction + else ([] if phase == "activity" else [Author]))), activities=[echo] if phase == "activity" else [], sticky_cache_capacity=0, sticky_cache_ttl_seconds=1) running = asyncio.create_task(worker.run()) - wanted = ({"original": "TimerScheduled", "replacement": "WorkflowCompleted"} if timer_author else + wanted = ({"original": "ConditionWaitOpened", "resumed": "ConditionWaitOpened", + "replacement": "WorkflowCompleted"} if signal_direction else + ({"original": "TimerScheduled", "replacement": "WorkflowCompleted"} if timer_author else {"original": "ChildWorkflowScheduled", "child": "WorkflowCompleted", "replacement": "WorkflowCompleted"} if child_direction else {"original": "ActivityScheduled", "activity": "ActivityCompleted", - "replacement": "WorkflowCompleted"})[phase] + "replacement": "WorkflowCompleted"}))[phase] try: while True: + if signal_direction and not run_id: + if WORKFLOW_INVOCATIONS == 0: + if running.done(): + await running + raise RuntimeError("Published worker stopped before its first workflow task") + await asyncio.sleep(.025) + continue + run_id = (await client.describe_workflow(workflow_id)).run_id history = await client.get_history(workflow_id, run_id, page_size=1000) - if any(event["event_type"] == wanted for event in history["events"]): + observed = sum(event["event_type"] == wanted for event in history["events"]) + required = 2 if signal_direction and phase == "resumed" else 1 + if observed >= required: break if running.done(): await running diff --git a/scripts/conformance/server-parity/sdk-directions/rust/src/main.rs b/scripts/conformance/server-parity/sdk-directions/rust/src/main.rs index fc892f0db..939b189e0 100644 --- a/scripts/conformance/server-parity/sdk-directions/rust/src/main.rs +++ b/scripts/conformance/server-parity/sdk-directions/rust/src/main.rs @@ -5,7 +5,8 @@ use std::sync::{ use std::time::{Duration, Instant}; use durable_workflow::{ - ActivityOptions, ChildWorkflowOptions, Client, Error, Result, Worker, SDK_VERSION, + ActivityOptions, ChildWorkflowOptions, Client, ConditionWaitOptions, Error, Result, Worker, + SDK_VERSION, }; use serde_json::{json, Value}; @@ -28,7 +29,10 @@ async fn main() -> Result<()> { .map_err(|error| Error::WorkerLoop(error.to_string()))?; let child_direction = fixture.get("sdk_child_direction").is_some(); let timer_author = fixture.get("sdk_timer_author").is_some(); - let phases: &[&str] = if timer_author { + let signal_direction = fixture.get("sdk_signal_direction").is_some(); + let phases: &[&str] = if signal_direction { + &["original", "signal-1", "resumed", "signal-2", "replacement"] + } else if timer_author { &["original", "replacement"] } else if child_direction { &["original", "child", "replacement"] @@ -38,7 +42,9 @@ async fn main() -> Result<()> { if !phases.contains(&phase.as_str()) { return Err(Error::WorkerLoop("Unknown SDK direction phase".into())); } - let queue = if timer_author { + let queue = if signal_direction { + "server-parity-sdk-signals" + } else if timer_author { "server-parity-sdk-timer" } else if child_direction { if phase == "child" { @@ -62,7 +68,20 @@ async fn main() -> Result<()> { .namespace("default") .token(std::env::var("DW_PARITY_TOKEN").ok()) .build()?; - let run_id = if phase == "original" { + if signal_direction && matches!(phase.as_str(), "signal-1" | "signal-2") { + let result = client + .signal_workflow(workflow_id, "payload", json!([fixture["signal_value"]])) + .await?; + println!( + "{}", + json!({"pid": std::process::id(), "language": "rust", + "sdk_version": SDK_VERSION.trim_start_matches("durable-workflow-rust/"), + "phase": phase, "workflow_id": workflow_id, "run_id": selected_run, + "signal_sent": true, "sender_finished": true, "client_result": result}) + ); + return Ok(()); + } + let mut run_id = if phase == "original" && !signal_direction { client .start_workflow( fixture["workflow_type"] @@ -84,7 +103,47 @@ async fn main() -> Result<()> { let mut worker = Worker::new(client.clone(), queue) .worker_id(&worker_id) .poll_timeout(Duration::from_secs(1)); - if timer_author { + if signal_direction { + let invocations = Arc::clone(&workflow_invocations); + worker.register_workflow( + "parity.v1.sdk_signals", + move |ctx, arguments: Value| { + invocations.fetch_add(1, Ordering::SeqCst); + async move { + // The ordinary positional API preserves this two-argument input. + // Published typed handlers accept one request value only. + let target = arguments + .as_array() + .and_then(|values| values.get(1)) + .and_then(Value::as_u64) + .and_then(|value| usize::try_from(value).ok()) + .ok_or_else(|| Error::WorkerLoop("No signal count argument".into()))?; + for index in 0..target { + let predicate_ctx = ctx.clone(); + ctx.wait_condition( + ConditionWaitOptions::new( + format!("payload:{index}"), + // SHA256 of the two authored predicate identities. + if index == 0 { + "sha256:b67992a5ee48704611dcf6a181c3ab94b69d0933c44525137210dd2467aec563" + } else { + "sha256:474b0d8e3a116beb308c37e8084f7f16a402f8142468144d34ff135bc5ef0910" + }, + ), + move || Ok(predicate_ctx.signals("payload")?.len() > index), + ) + .await?; + } + ctx.signals("payload")? + .last() + .and_then(|arguments| arguments.first()) + .cloned() + .ok_or_else(|| Error::WorkerLoop("No final decoded signal value".into())) + } + }, + ); + worker.declare_workflow_signals("parity.v1.sdk_signals", &["payload"])?; + } else if timer_author { let invocations = Arc::clone(&workflow_invocations); worker.register_typed_workflow("parity.v1.sdk_timer", move |ctx, value: Value| { invocations.fetch_add(1, Ordering::SeqCst); @@ -154,8 +213,22 @@ async fn main() -> Result<()> { } // Public SDK registration and task loop: no synthetic protocol or history. worker.run_once().await?; + if signal_direction && run_id.is_empty() { + if workflow_invocations.load(Ordering::SeqCst) == 0 { + continue; + } + run_id = client + .describe_workflow(workflow_id) + .await? + .run_id + .ok_or_else(|| Error::WorkerLoop("No original claimed run ID".into()))?; + } let run = client.describe_workflow_run(workflow_id, &run_id).await?; let reached = match phase.as_str() { + "original" | "resumed" if signal_direction => { + workflow_invocations.load(Ordering::SeqCst) == 1 + && run.status.as_deref() == Some("waiting") + } "original" => run.status.as_deref() == Some("waiting"), "activity" => activity_invocations.load(Ordering::SeqCst) == 1, _ => run.status.as_deref() == Some("completed"), diff --git a/scripts/conformance/server-parity/sdk-signal-direction-contract.mjs b/scripts/conformance/server-parity/sdk-signal-direction-contract.mjs new file mode 100644 index 000000000..541db2c5b --- /dev/null +++ b/scripts/conformance/server-parity/sdk-signal-direction-contract.mjs @@ -0,0 +1,183 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkPublishedSdkPackages} from './sdk-direction-contract.mjs'; + +const artifacts = JSON.parse(readFileSync(new URL('../../../tests/Fixtures/ServerParityProfiles/published-sdk-signals/artifacts.json', import.meta.url), 'utf8')); +const languages = ['php', 'python', 'rust']; +const phases = ['original', 'signal-1', 'resumed', 'signal-2', 'replacement']; +const checkpointLengths = [3, 4, 8, 9, 13]; +const claimLengths = [2, 4, 9]; +const nonempty = value => assert(typeof value === 'string' && value.length > 0); +const history = events => events.map(({sequence, event_type, payload}) => ({sequence, event_type, payload})); +const frame = value => { + if (typeof value === 'string') { nonempty(value); return value; } + assert.equal(value?.codec, 'avro'); nonempty(value?.blob); return value.blob; +}; + +export function checkSdkSignalDirectionArtifacts(fixture, selected) { + if (!fixture.sdk_signal_direction) return; + assert.deepEqual(selected, artifacts, 'complete reviewed published SDK signal tuple'); + const spec = fixture.sdk_signal_direction; + assert(languages.includes(spec.workflow) && languages.includes(spec.client)); + assert.deepEqual(spec, {workflow: spec.workflow, client: spec.client, phase_order: phases, + task_queue: 'server-parity-sdk-signals', signal_name: 'payload', target: 'instance-current-run', + start_after_worker_registration: true, first_poll_after_start_receipt: true, + start_client: 'php-observer', identical_repeated_payload: true}); + assert.equal(fixture.id, 'signal-sdk-' + spec.workflow + '-' + spec.client); + assert.equal(fixture.workflow_type, 'parity.v1.sdk_signals'); + assert.equal(fixture.signal_count, 2); + assert.equal(fixture.activity, false); + assert.equal(fixture.typed_signal_value.value.count.value, '9007199254740993'); +} + +export function checkSdkSignalDirectionPackages(fixture, observation, selected) { + if (!fixture.sdk_signal_direction) return; + checkSdkSignalDirectionArtifacts(fixture, selected); + checkPublishedSdkPackages(observation, selected, observation.sdk_signal_direction); +} + +export function checkSdkSignalDirection(fixture, observation, workflowId) { + if (!fixture.sdk_signal_direction) return null; + checkSdkSignalDirectionArtifacts(fixture, artifacts); + const spec = fixture.sdk_signal_direction, direction = observation.sdk_signal_direction; + const projection = {two_distinct_identical_signals: true, exact_typed_completion: true}; + if (observation.mode === 'embedded') { + assert.deepEqual(direction, {applicable: false, reason: 'embedded_executes_php_author_definitions'}); + return projection; + } + assert.equal(direction.applicable, true); + assert.deepEqual(direction.phases.map(item => item.phase), phases); + assert.equal(new Set(direction.phases.map(item => item.pid)).size, 5, 'five distinct actual SDK processes'); + const observer = direction.observer_start; + assert.equal(observer.role, 'observer'); + assert.equal(observer.language, 'php'); + assert.equal(observer.sdk_version, artifacts.sdk_php); + assert(Number.isInteger(observer.pid) && observer.pid > 0); + assert(!direction.phases.some(item => item.pid === observer.pid), 'observer is separate from all SDK phases'); + assert.equal(observer.run_id, observation.run_id); + const claims = [], completions = [], signals = [], starts = []; + let turn = 0; + for (const [index, phase] of direction.phases.entries()) { + const sender = index === 1 || index === 3; + assert.equal(phase.language, sender ? spec.client : spec.workflow); + assert.equal(phase.sdk_version, artifacts['sdk_' + phase.language]); + assert(Number.isInteger(phase.pid) && phase.pid > 0); + assert.equal(phase.run_id, observation.run_id); + assert.equal(phase.worker_error, undefined); + assert.equal(phase.observer_error, undefined); + const snapshot = phase.snapshot; + for (const key of ['workflow_id', 'run_id', 'workflow_type', 'namespace', 'task_queue', 'payload_codec']) { + assert.equal(snapshot[key], observation[key], 'original run ' + key + ' throughout the phases'); + } + assert.equal(snapshot.events.length, checkpointLengths[index]); + assert.deepEqual(snapshot.events, observation.events.slice(0, checkpointLengths[index]), 'complete unchanged original history prefix'); + assert.deepEqual(snapshot.typed_input, observation.typed_input); + assert.deepEqual(snapshot.typed_output, index === 4 ? fixture.typed_signal_value : {type: 'null', value: null}); + if (sender) assert(['waiting', 'running'].includes(snapshot.status), 'accepted signal leaves original run nonterminal'); + else assert.equal(snapshot.status, index === 4 ? 'completed' : 'waiting'); + for (const request of phase.requests) if (request.method === 'POST' && request.path === '/api/workflows') starts.push({index, request}); + if (sender) { + assert.equal(phase.sender_finished, true); + assert.equal(phase.signal_sent, true); + assert.equal(phase.workflow_id, workflowId); + assert(!phase.requests.some(item => item.path.startsWith('/api/worker/')), 'normal signal client performs no worker operations'); + const requests = phase.requests.filter(item => item.method === 'POST' && item.path === '/api/workflows/' + workflowId + '/signal/payload'); + assert.equal(requests.length, 1, 'one actual instance-scoped signal request per client process'); + const request = requests[0], receipt = request.response; + assert.equal(request.status, 202, 'normal accepted signal uses the published HTTP 202 contract'); + assert.equal(request.request.run_id, undefined, 'normal common instance API has no run selector'); + frame(request.request.input); + assert.deepEqual(request.decoded_input, [fixture.signal_value]); + assert.deepEqual(request.typed_decoded_input, {type: 'list', value: [fixture.typed_signal_value]}); + assert.equal(receipt.command_status ?? (receipt.accepted ? 'accepted' : 'rejected'), 'accepted'); + assert.equal(receipt.outcome, 'signal_received'); + assert.equal(receipt.workflow_id, workflowId); + assert.equal(receipt.run_id, observation.run_id); + nonempty(receipt.command_id); + if (spec.client === 'python') assert.equal(phase.client_result, null, 'published Python has no client receipt object'); + else assert.deepEqual(phase.client_result, receipt, 'retain the actual published client result'); + const delivery = observation.signal_deliveries[signals.length]; + assert.deepEqual(delivery.response, receipt, 'bind delivered signal to its actual receipt'); + assert.deepEqual(delivery.before, direction.phases[index - 1].snapshot.execution); + assert.equal(observation.events[index === 1 ? 3 : 8].payload.workflow_command_id, receipt.command_id); + signals.push(receipt.command_id); + continue; + } + assert.equal(phase.worker_finished, true); + assert.equal(phase.worker_id, workflowId + ':' + phase.phase); + assert.equal(phase.task_queue, spec.task_queue); + assert.equal(phase.workflow_invocations, 1); + assert.equal(phase.activity_invocations, 0); + const registrations = phase.requests.filter(item => item.method === 'POST' && item.path === '/api/worker/register'); + assert.equal(registrations.length, 1); + const registration = registrations[0]; + assert.equal(registration.status, 201); + assert.equal(registration.response.registered, true); + assert.equal(registration.request.worker_id, phase.worker_id); + assert.equal(registration.request.task_queue, spec.task_queue); + assert.deepEqual(registration.request.supported_workflow_types, [fixture.workflow_type]); + assert.deepEqual(registration.request.supported_activity_types, []); + const polls = phase.requests.filter(item => item.path === '/api/worker/workflow-tasks/poll' && item.response?.task); + assert.equal(polls.length, 1); + assert.equal(polls[0].method, 'POST'); assert.equal(polls[0].status, 200); + assert.equal(polls[0].request.task_queue, spec.task_queue); + const task = polls[0].response.task; + for (const key of ['workflow_id', 'run_id', 'workflow_type', 'task_queue']) assert.equal(task[key], observation[key]); + nonempty(task.task_id); nonempty(task.lease_owner); + assert.equal(task.workflow_task_attempt, 1); + assert.deepEqual(history(task.history_events), history(observation.events.slice(0, claimLengths[turn]))); + claims.push(task); + const outcomes = phase.requests.filter(item => item.method === 'POST' && /^\/api\/worker\/workflow-tasks\/[^/]+\/complete$/.test(item.path)); + assert.equal(outcomes.length, 1); + const outcome = outcomes[0]; + assert.equal(outcome.status, 200); assert.equal(outcome.response.recorded, true); + assert.equal(outcome.response.task_id, task.task_id); + assert.equal(outcome.response.run_id, observation.run_id); + assert.equal(outcome.path, '/api/worker/workflow-tasks/' + task.task_id + '/complete'); + assert.equal(outcome.request.lease_owner, task.lease_owner); + assert.equal(outcome.request.workflow_task_attempt, task.workflow_task_attempt); + assert.equal(outcome.request.commands.length, 1); + const command = outcome.request.commands[0]; + assert.equal(command.type, turn === 2 ? 'complete_workflow' : 'open_condition_wait'); + if (turn < 2) { + const opened = observation.events[turn === 0 ? 2 : 7].payload; + assert.equal(command.condition_key, 'payload:' + turn); + assert.equal(command.condition_definition_fingerprint, opened.condition_definition_fingerprint); + if (spec.workflow === 'rust') { + const occurrence = 'rust:condition-wait:' + turn; + assert.equal(command.condition_wait_occurrence_id, occurrence, 'actual authored Rust condition occurrence'); + assert.equal(opened.condition_wait_occurrence_id, occurrence, 'opening retains authored occurrence'); + assert.equal(observation.events[turn === 0 ? 6 : 11].payload.condition_wait_occurrence_id, + occurrence, 'matching satisfaction retains the original authored occurrence'); + } + } + assert(!phase.requests.some(item => item.path === '/api/worker/activity-tasks/poll' && item.response?.task)); + completions.push(outcome); turn++; + } + assert.equal(new Set(signals).size, 2, 'identical values are two distinct accepted signal commands'); + assert.equal(new Set(claims.map(item => item.task_id)).size, 3); + assert.deepEqual(observation.workflow_polls, claims, 'bind flattened claims to their real phase receipts'); + assert.equal(observation.workflow_completions.length, 3); + for (const [index, outcome] of completions.entries()) { + const {decoded_commands, ...wire} = observation.workflow_completions[index]; + assert.deepEqual(wire, outcome, 'bind flattened completions to real phase receipts'); + assert.equal(decoded_commands.length, 1); + if (index === 2) { + assert.deepEqual(decoded_commands[0].decoded.result, fixture.signal_value); + assert.deepEqual(decoded_commands[0].typed_decoded.result, fixture.typed_signal_value); + assert.equal(frame(outcome.request.commands[0].result), frame(observation.execution.output_envelope)); + } else { + assert.deepEqual(decoded_commands[0], {decoded: {}, typed_decoded: {}}); + } + } + assert.equal(starts.length, 1, 'no cold restart creates another run'); + assert.equal(starts[0].index, 0); + assert.equal(starts[0].request.status, 201); + assert.equal(starts[0].request.response.run_id, observation.run_id); + for (const key of ['workflow_id', 'workflow_type', 'task_queue']) assert.equal(starts[0].request.request[key], observation[key]); + assert(direction.phases[0].requests.findIndex(item => item.path === '/api/worker/workflow-tasks/poll' && item.response?.task) + > direction.phases[0].requests.indexOf(starts[0].request), 'original workflow claim follows the real accepted start receipt'); + assert(direction.phases[0].requests.indexOf(starts[0].request) > + direction.phases[0].requests.findIndex(item => item.path === '/api/worker/register'), 'start follows the actual successful registration'); + return projection; +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/artifacts.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/artifacts.json new file mode 100644 index 000000000..e32ea7520 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/artifacts.json @@ -0,0 +1,68 @@ +{ + "schema": "durable-workflow.server-port-artifacts/v1", + "server": { + "version": "2.5.20", + "source_commit": "fdce11f2b3da1b5ef3e6ad2e081a65f02f28019d", + "workflow": "2.5.10", + "workflow_source_commit": "52e810c7971d0af9907f2d59b349efedd43e7749", + "image_index": "sha256:38e7cb9c4f6756a4dbb23fc933858312fe5845870b991a101d7545193ee8e6ee", + "images": { + "linux/amd64": "durableworkflow/server@sha256:d4a146f9054bbbc47f060734cca9b1fcbeb7834ed77e0dc1e9a9bd4a188af862", + "linux/arm64": "durableworkflow/server@sha256:31bce3f00cbdb070fc4548fa290a4a3c6c7e77805af3d6d31dbbb5294d021cc0" + } + }, + "workflow": "2.5.10", + "workflow_source_commit": "52e810c7971d0af9907f2d59b349efedd43e7749", + "workflow_loaded_sources": { + "src/V2/Support/TaskRepair.php": "744339f862de8302cbeddee74f4c1a31881582a1649e73f63f952c03ae2d0ecc", + "src/V2/Support/VersionResolver.php": "473b2ecd645d688622735508dd4460cff95e32ad17d25a6e3206e781cee0adf7", + "src/V2/Support/WorkflowExecutor.php": "9b85a1daf80fbe63d4f0ec3e23513c213e4971379f9a280495c0df6dc9655062" + }, + "embedded_image_composer_lock_sha256": "9c6195e596c430c958e68ac375c236fcee7c4d1dc9929ed44311d266cc62ce0d", + "embedded_image_path_reference": "09f6dde2edfc612addea5a134c9183d8062deb5e", + "sdk_php": "2.2.7", + "sdk_php_source_commit": "7c40c7755789790ce4c0ce706a8e3a6462742956", + "sdk_php_composer_lock_sha256": "af600c49741d714aea69e2830309db8065461c9fd6b0e499c68699f74641faab", + "sdk_php_loaded_sources": { + "DurableWorkflow\\Worker": "38722c5de194037ae49f610f68436ff11e82de2c5ca0d2c7ac2f66fce1972983", + "DurableWorkflow\\Worker\\WorkflowContext": "ba8dbc9ac535490109862270e1f734dfdd83fda065ae1aeada3799b156f99d2f" + }, + "sdk_python": "2.5.4", + "sdk_rust": "3.4.4", + "cli": "2.2.3", + "waterline": "2.9.1", + "capacity_suite": "1.9.0", + "performance_status": "not_measured", + "qualification_scope": "Published PHP/Python/Rust instance-scoped signal clients against each PHP/Python/Rust workflow author; first workflow poll after the successful start receipt, two distinct identical accepted signals, original run/queue/history and three cold author processes; canonical embedded counterparts", + "resource_limits": { + "cpus": 1, + "memory_bytes": 536870912, + "memory_swap_bytes": 536870912 + }, + "database_memory_limits_bytes": { + "mysql-8.0": 1073741824, + "existing-php-mysql-matrix": 1073741824, + "mariadb-10.11": 536870912, + "postgres-17": 536870912, + "existing-php-matrix": 536870912 + }, + "runtime_targets": [ + "php", + "rust", + "embedded" + ], + "published_sdk_artifacts": { + "python": { + "version": "2.5.4", + "source_commit": "86cfced48a3938b41d707f26eea1ff098e07ffcb", + "archive_sha256": "55e1be185f5bbe6e8c6dac10c36d5a777d11d85e7bf052eed1926cc02100889d" + }, + "rust": { + "version": "3.4.4", + "source_commit": "6f316506de136ebaa7124cdfd011f656b355192a", + "archive_sha256": "b3e0acde01baa1e552735af89a5623bee37b90f62405105abca480fc978874d6" + } + }, + "prior_published_tuples_unchanged": true, + "resource_scope": "Current conformance CI allowances; historical receipts retain their original limits and this profile does not authorize a performance comparison" +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-php.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-php.json new file mode 100644 index 000000000..abd5effca --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-php.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-php-php", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "php", + "client": "php", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-python.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-python.json new file mode 100644 index 000000000..538345d92 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-python.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-php-python", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "php", + "client": "python", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-rust.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-rust.json new file mode 100644 index 000000000..f1b96ea0b --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-php-rust.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-php-rust", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "php", + "client": "rust", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-php.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-php.json new file mode 100644 index 000000000..05193ce3b --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-php.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-python-php", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "python", + "client": "php", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-python.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-python.json new file mode 100644 index 000000000..3e16c7be0 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-python.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-python-python", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "python", + "client": "python", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-rust.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-rust.json new file mode 100644 index 000000000..d98fc5606 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-python-rust.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-python-rust", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "python", + "client": "rust", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-php.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-php.json new file mode 100644 index 000000000..51a3ee993 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-php.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-rust-php", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "rust", + "client": "php", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-python.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-python.json new file mode 100644 index 000000000..716e3dd6d --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-python.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-rust-python", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "rust", + "client": "python", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-rust.json b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-rust.json new file mode 100644 index 000000000..3f57abc96 --- /dev/null +++ b/tests/Fixtures/ServerParityProfiles/published-sdk-signals/signal-sdk-rust-rust.json @@ -0,0 +1,93 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "signal-sdk-rust-rust", + "workflow_type": "parity.v1.sdk_signals", + "input": { + "message": "signal workflow λ" + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "signal workflow λ" + } + } + }, + "signal_value": { + "message": "cold signal λ", + "count": 9007199254740993, + "ready": true, + "optional": null + }, + "typed_signal_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cold signal λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "ready": { + "type": "boolean", + "value": true + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "signal_count": 2, + "sdk_signal_direction": { + "workflow": "rust", + "client": "rust", + "phase_order": [ + "original", + "signal-1", + "resumed", + "signal-2", + "replacement" + ], + "task_queue": "server-parity-sdk-signals", + "signal_name": "payload", + "target": "instance-current-run", + "start_after_worker_registration": true, + "first_poll_after_start_receipt": true, + "start_client": "php-observer", + "identical_repeated_payload": true + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "ConditionWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "ConditionWaitSatisfied", + "WorkflowCompleted" + ], + "embedded_expected_events": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "SignalWaitOpened", + "SignalReceived", + "MessageCursorAdvanced", + "SignalApplied", + "WorkflowCompleted" + ], + "contract": "The normal SDK worker registers its signal declaration before the observer starts the original run, and the first real workflow poll is forwarded only after the matching successful start response is observed. Two distinct normal published client processes send identical instance-scoped payloads to that current run. Three distinct author processes open the first condition wait, cold-replay the first signal into a second wait, then cold-replay both accepted signals and complete the same original run with the exact final signal payload. Preserve distinct command/signal identities, ordered acceptance/application and cursor advancement, original waits/queue/history, actual registration and fenced receipts, and exact int64/Unicode/boolean/null Avro values. Embedded uses its canonical direct-signal author; its explicitly different full history is retained and SDK language transport is inapplicable. The retained PHP startup-overlap failure is a separate unqualified interleaving. No continuation/current-run race, run-targeted client API, all signal policies or performance claim." +} diff --git a/tests/Unit/ServerParityPublishedPatchTransportTest.mjs b/tests/Unit/ServerParityPublishedPatchTransportTest.mjs index c2d54af7f..3284cea09 100644 --- a/tests/Unit/ServerParityPublishedPatchTransportTest.mjs +++ b/tests/Unit/ServerParityPublishedPatchTransportTest.mjs @@ -10,6 +10,115 @@ import {join} from 'node:path'; import {gzipSync} from 'node:zlib'; import {setTimeout as delay} from 'node:timers/promises'; +async function startGateRequest(url, body) { + return new Promise((resolve, reject) => { + const request = http.request(url, {method: 'POST', headers: {'content-type': 'application/json'}}, response => { + const chunks = []; + response.on('data', bytes => chunks.push(bytes)); + response.on('end', () => resolve({status: response.statusCode, bytes: Buffer.concat(chunks)})); + response.on('error', reject); + }); + request.on('error', reject); + request.end(body); + }); +} + +test('start gate forwards original poll bytes only after the matching successful start receipt', {timeout: 10000}, async () => { + const directory = mkdtempSync(join(tmpdir(), 'server-parity-start-gate-')); + const journal = join(directory, 'wire.jsonl'); + const pollBody = '{"worker_id":"original-worker","timeout_seconds":1,"count":9007199254740993}'; + const pollReply = '{"task":{"task_id":"original-task","value":9007199254740993}}'; + let polls = 0, accepted = false, observedPoll; + const upstream = http.createServer((request, response) => { + const chunks = []; + request.on('data', bytes => chunks.push(bytes)); + request.on('end', () => { + const body = Buffer.concat(chunks).toString(); + if (request.url === '/api/worker/workflow-tasks/poll') { + polls++; + observedPoll = {body, accepted}; + response.end(pollReply); + } else { + const input = JSON.parse(body); + setTimeout(() => { + const success = input.workflow_id !== 'original-workflow' || input.accept === true; + if (input.workflow_id === 'original-workflow' && success) accepted = true; + response.writeHead(success ? 201 : 422); + response.end(JSON.stringify({run_id: success ? input.workflow_id + ':run' : null})); + }, 30); + } + }); + }); + upstream.listen(0, '127.0.0.1'); await once(upstream, 'listening'); + const proxy = spawn(process.execPath, [new URL('../../scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs', import.meta.url).pathname, + `http://127.0.0.1:${upstream.address().port}`, journal, '--forwarding-state', '--workflow-poll-after-start', 'original-workflow'], + {stdio: ['ignore', 'pipe', 'pipe']}); + try { + const [ready] = await once(proxy.stdout, 'data'); const url = ready.toString().trim(); + const pending = startGateRequest(url + '/api/worker/workflow-tasks/poll', pollBody); + assert.equal((await startGateRequest(url + '/api/workflows', '{"workflow_id":"other-workflow"}')).status, 201); + assert.equal(polls, 0, 'another workflow cannot release the original poll'); + assert.equal((await startGateRequest(url + '/api/workflows', '{"workflow_id":"original-workflow","accept":false}')).status, 422); + assert.equal(polls, 0, 'a rejected start cannot release the original poll'); + const started = await startGateRequest(url + '/api/workflows', '{"workflow_id":"original-workflow","accept":true}'); + assert.equal(started.status, 201); + const result = await pending; + assert.deepEqual(observedPoll, {body: pollBody, accepted: true}); + assert.equal(result.status, 200); assert.equal(result.bytes.toString(), pollReply); assert.equal(polls, 1); + const wire = readFileSync(journal, 'utf8').trim().split('\n').map(JSON.parse); + assert.equal(wire.length, 4); + assert.equal(wire[2].path, '/api/workflows'); assert.equal(wire[2].status, 201); + assert.equal(wire[3].path, '/api/worker/workflow-tasks/poll'); + assert.equal(wire[3].request_body, pollBody); assert.equal(wire[3].response_body, pollReply); + assert.equal(wire[3].upstream_request_started, true); assert.equal(wire[3].request_body_complete, true); + } finally { + const closed = once(proxy, 'close'); proxy.kill(); await closed; + upstream.closeAllConnections(); await new Promise(resolve => upstream.close(resolve)); + rmSync(directory, {recursive: true}); + } +}); + +test('cancelled fully received gated poll never reaches upstream or receives an invented claim', {timeout: 10000}, async () => { + const directory = mkdtempSync(join(tmpdir(), 'server-parity-start-cancel-')); + const journal = join(directory, 'wire.jsonl'); + let polls = 0; + const upstream = http.createServer((request, response) => { + if (request.url === '/api/worker/workflow-tasks/poll') polls++; + request.resume(); request.on('end', () => {response.writeHead(201); response.end('{"run_id":"original-run"}');}); + }); + upstream.listen(0, '127.0.0.1'); await once(upstream, 'listening'); + const proxy = spawn(process.execPath, [new URL('../../scripts/conformance/server-parity/published-sdk-patch/observe-http.mjs', import.meta.url).pathname, + `http://127.0.0.1:${upstream.address().port}`, journal, '--forwarding-state', '--workflow-poll-after-start', 'original-workflow'], + {stdio: ['ignore', 'pipe', 'pipe']}); + let client; + try { + const [ready] = await once(proxy.stdout, 'data'); const url = ready.toString().trim(); + client = http.request(url + '/api/worker/workflow-tasks/poll', {method: 'POST'}); + client.on('error', () => {}); client.end('{"worker_id":"cancelled-original"}'); + await startGateRequest(url + '/api/workflows', '{"workflow_id":"other-workflow"}'); + assert.equal(polls, 0); client.destroy(); + const deadline = Date.now() + 2000; + let receipt; + while (!receipt) { + const rows = readFileSync(journal, 'utf8').trim().split('\n').map(JSON.parse); + receipt = rows.find(item => item.path === '/api/worker/workflow-tasks/poll'); + if (!receipt) {assert(Date.now() < deadline); await delay(10);} + } + assert.equal(receipt.status, 0); assert.equal(receipt.client_cancelled, true); + assert.equal(receipt.request_body_complete, true); assert.equal(receipt.upstream_request_started, false); + assert.equal(receipt.request_body, '{"worker_id":"cancelled-original"}'); + assert.equal(receipt.response_body, ''); assert.equal(receipt.response_bytes_base64, ''); + await startGateRequest(url + '/api/workflows', '{"workflow_id":"original-workflow"}'); + assert.equal(polls, 0, 'accepting the start cannot resurrect the cancelled poll'); + } finally { + client?.destroy(); + const closed = once(proxy, 'close'); proxy.kill(); await closed; + upstream.closeAllConnections(); await new Promise(resolve => upstream.close(resolve)); + rmSync(directory, {recursive: true}); + } +}); + + test('HTTP observer preserves compressed bytes and refuses to retain authentication headers', {timeout: 10000}, async () => { const directory = mkdtempSync(join(tmpdir(), 'server-parity-wire-')); const journal = join(directory, 'wire.jsonl'); diff --git a/tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs b/tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs new file mode 100644 index 000000000..ba9d63af6 --- /dev/null +++ b/tests/Unit/ServerParitySdkSignalDirectionContractTest.mjs @@ -0,0 +1,201 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkObservation} from '../../scripts/conformance/server-parity/contract.mjs'; +import {checkSdkSignalDirectionArtifacts, checkSdkSignalDirectionPackages} from '../../scripts/conformance/server-parity/sdk-signal-direction-contract.mjs'; + +const profile = new URL('../Fixtures/ServerParityProfiles/published-sdk-signals/', import.meta.url); +const artifacts = JSON.parse(readFileSync(new URL('artifacts.json', profile), 'utf8')); +const fixtureFor = (author, client) => JSON.parse(readFileSync(new URL('signal-sdk-' + author + '-' + client + '.json', profile), 'utf8')); +const copy = value => structuredClone(value); +const workflowId = 'signal-original-workflow'; +const phases = ['original', 'signal-1', 'resumed', 'signal-2', 'replacement']; +const lengths = [3, 4, 8, 9, 13]; +const nil = {type: 'null', value: null}; + +function observed(fixture, mode = 'http') { + const spec = fixture.sdk_signal_direction; + const raw = {mode, workflow_id: workflowId, run_id: 'original-signal-run', workflow_type: fixture.workflow_type, + namespace: 'default', task_queue: spec.task_queue, status: 'completed', payload_codec: 'avro', + input: [fixture.input, 2], typed_input: {type: 'list', value: [fixture.typed_value, {type: 'int64', value: '2'}]}, + output: fixture.signal_value, typed_output: fixture.typed_signal_value, + execution: {run_id: 'original-signal-run', status: 'completed', started_at: '2026-01-01T00:00:00Z', + output_envelope: {codec: 'avro', blob: 'original-output-frame'}}, + sdk_php: artifacts.sdk_php, sdk_php_source: artifacts.sdk_php_source_commit, + sdk_composer_lock_sha256: artifacts.sdk_php_composer_lock_sha256, + sdk_loaded_sources: copy(artifacts.sdk_php_loaded_sources)}; + const start = {workflow_instance_id: workflowId, workflow_run_id: raw.run_id, + workflow_command_id: 'original-start-command', workflow_type: fixture.workflow_type}; + raw.events = [ + {event_type: 'StartAccepted', payload: {...start, outcome: 'started_new'}, decoded: {}, typed_decoded: {}}, + {event_type: 'WorkflowStarted', payload: {...start, declared_signals: ['payload'], + execution_timeout_seconds: 3600, run_timeout_seconds: 600, + execution_deadline_at: '2026-01-01T01:00:00Z', run_deadline_at: '2026-01-01T00:10:00Z'}, decoded: {}, typed_decoded: {}}, + ]; + raw.signal_deliveries = []; + for (let index = 0; index < 2; index++) { + const opened = mode === 'embedded' ? {signal_wait_id: 'wait-' + index, signal_name: 'payload', sequence: index + 1} + : {condition_wait_id: 'wait-' + index, condition_key: 'payload:' + index, + ...(spec.workflow === 'rust' ? {condition_wait_occurrence_id: 'rust:condition-wait:' + index} : {}), + condition_definition_fingerprint: 'sha256:' + 'a'.repeat(64), sequence: index + 1}; + const received = {workflow_instance_id: workflowId, workflow_run_id: raw.run_id, + workflow_command_id: 'signal-command-' + index, signal_id: 'signal-' + index, + signal_wait_id: (mode === 'embedded' ? 'wait-' : 'routing-wait-') + index, + signal_name: 'payload', payload_codec: 'avro', + command: {id: 'signal-command-' + index, sequence: index + 2, message_sequence: index + 1}}; + const receipt = {accepted: true, command_id: received.workflow_command_id, run_id: raw.run_id, + workflow_id: workflowId, outcome: 'signal_received'}; + raw.signal_deliveries.push({before: {status: 'waiting', run_id: raw.run_id}, response: receipt}); + raw.events.push( + {event_type: mode === 'embedded' ? 'SignalWaitOpened' : 'ConditionWaitOpened', payload: opened, decoded: {}, typed_decoded: {}}, + {event_type: 'SignalReceived', payload: received, decoded: {arguments: [fixture.signal_value]}, + typed_decoded: {arguments: {type: 'list', value: [fixture.typed_signal_value]}}}, + {event_type: 'MessageCursorAdvanced', payload: {stream_key: 'instance:' + workflowId, + previous_position: index, new_position: index + 1}, decoded: {}, typed_decoded: {}}, + {event_type: 'SignalApplied', payload: {...received, ...(mode === 'embedded' ? {sequence: index + 1} : {})}, + decoded: {value: fixture.signal_value}, typed_decoded: {value: fixture.typed_signal_value}}, + ...(mode === 'embedded' ? [] : [{event_type: 'ConditionWaitSatisfied', payload: {...opened, + workflow_signal_id: received.signal_id, signal_name: received.signal_name, signal_wait_id: received.signal_wait_id}, + decoded: {}, typed_decoded: {}}])); + } + raw.events.push({event_type: 'WorkflowCompleted', payload: {}, decoded: {output: raw.output}, typed_decoded: {output: raw.typed_output}}); + raw.events = raw.events.map((event, index) => ({...event, sequence: index + 1, timestamp: '2026-01-01T00:00:01Z'})); + if (mode === 'embedded') { + Object.assign(raw, {workflow_package: artifacts.workflow, workflow_source: artifacts.embedded_image_path_reference, + embedded_composer_lock_sha256: artifacts.embedded_image_composer_lock_sha256, + workflow_loaded_sources: copy(artifacts.workflow_loaded_sources), + sdk_signal_direction: {applicable: false, reason: 'embedded_executes_php_author_definitions'}}); + return copy(raw); + } + raw.workflow_polls = []; raw.workflow_completions = []; + const recorded = []; + let turn = 0; + const request = (path, body, response, status = 200) => ({method: 'POST', path, status, request: body, response}); + for (const [index, phase] of phases.entries()) { + const sender = index === 1 || index === 3; + const checkpoint = {workflow_id: workflowId, run_id: raw.run_id, workflow_type: raw.workflow_type, + namespace: raw.namespace, task_queue: raw.task_queue, payload_codec: raw.payload_codec, + status: index === 4 ? 'completed' : 'waiting', events: copy(raw.events.slice(0, lengths[index])), + typed_input: copy(raw.typed_input), typed_output: copy(index === 4 ? raw.typed_output : nil), + execution: {...raw.execution, status: index === 4 ? 'completed' : 'waiting', + output_envelope: index === 4 ? raw.execution.output_envelope : null}}; + if (sender) { + const signalIndex = index === 1 ? 0 : 1, receipt = copy(raw.signal_deliveries[signalIndex].response); + raw.signal_deliveries[signalIndex].before = copy(recorded[index - 1].snapshot.execution); + const wire = request('/api/workflows/' + workflowId + '/signal/payload', + {input: {codec: 'avro', blob: 'signal-frame-' + signalIndex}}, receipt, 202); + wire.decoded_input = [fixture.signal_value]; + wire.typed_decoded_input = {type: 'list', value: [fixture.typed_signal_value]}; + recorded.push({phase, language: spec.client, sdk_version: artifacts['sdk_' + spec.client], + pid: 201 + index, workflow_id: workflowId, run_id: raw.run_id, sender_finished: true, + signal_sent: true, client_result: spec.client === 'python' ? null : copy(receipt), requests: [wire], snapshot: checkpoint}); + continue; + } + const workerId = workflowId + ':' + phase; + const task = {task_id: 'task-' + phase, lease_owner: workerId, workflow_task_attempt: 1, + workflow_id: workflowId, run_id: raw.run_id, workflow_type: raw.workflow_type, task_queue: raw.task_queue, + history_events: copy(raw.events.slice(0, [2, 4, 9][turn]))}; + const command = turn === 2 ? {type: 'complete_workflow', result: raw.execution.output_envelope} + : {type: 'open_condition_wait', condition_key: 'payload:' + turn, + ...(spec.workflow === 'rust' ? {condition_wait_occurrence_id: 'rust:condition-wait:' + turn} : {}), + condition_definition_fingerprint: raw.events[turn === 0 ? 2 : 7].payload.condition_definition_fingerprint}; + const requests = [request('/api/worker/register', {worker_id: workerId, task_queue: raw.task_queue, + supported_workflow_types: [raw.workflow_type], supported_activity_types: []}, {registered: true}, 201)]; + if (turn === 0) requests.push(request('/api/workflows', {workflow_id: workflowId, + workflow_type: raw.workflow_type, task_queue: raw.task_queue, input: {codec: 'avro', blob: 'input-frame'}}, {run_id: raw.run_id}, 201)); + requests.push(request('/api/worker/workflow-tasks/poll', {task_queue: raw.task_queue}, {task})); + const completion = request('/api/worker/workflow-tasks/' + task.task_id + '/complete', + {lease_owner: task.lease_owner, workflow_task_attempt: 1, commands: [command]}, + {recorded: true, task_id: task.task_id, run_id: raw.run_id}); + requests.push(completion); + raw.workflow_polls.push(copy(task)); + raw.workflow_completions.push({...copy(completion), decoded_commands: [turn === 2 + ? {decoded: {result: raw.output}, typed_decoded: {result: raw.typed_output}} + : {decoded: {}, typed_decoded: {}}]}); + recorded.push({phase, language: spec.workflow, sdk_version: artifacts['sdk_' + spec.workflow], + pid: 201 + index, worker_finished: true, run_id: raw.run_id, worker_id: workerId, + task_queue: raw.task_queue, workflow_invocations: 1, activity_invocations: 0, requests, snapshot: checkpoint}); + turn++; + } + raw.sdk_signal_direction = {applicable: true, phases: recorded, + observer_start: {role: 'observer', language: 'php', sdk_version: artifacts.sdk_php, pid: 120, run_id: raw.run_id}, + packages: {outcome: 'pass', rust: copy(artifacts.published_sdk_artifacts.rust), python: copy(artifacts.published_sdk_artifacts.python)}}; + return copy(raw); +} + +for (const author of ['php', 'python', 'rust']) for (const client of ['php', 'python', 'rust']) for (const mode of ['http', 'embedded']) { + test('check signal direction guard model ' + author + '/' + client + ' ' + mode, () => { + const fixture = fixtureFor(author, client), value = observed(fixture, mode); + checkSdkSignalDirectionPackages(fixture, value, artifacts); + assert.deepEqual(checkObservation(fixture, value, workflowId).sdk_signal_direction, + {two_distinct_identical_signals: true, exact_typed_completion: true}); + }); +} +const fixture = fixtureFor('python', 'python'); +const replacement = value => value.sdk_signal_direction.phases[4]; +const firstSender = value => value.sdk_signal_direction.phases[1]; +const completion = value => replacement(value).requests.at(-1); +const mutations = { + 'same process replay': value => {replacement(value).pid = value.sdk_signal_direction.phases[0].pid;}, + 'missing cold process': value => {value.sdk_signal_direction.phases.pop();}, + 'observer is SDK worker': value => {value.sdk_signal_direction.observer_start.pid = replacement(value).pid;}, + 'wrong sender language': value => {firstSender(value).language = 'rust';}, + 'wrong installed version': value => {replacement(value).sdk_version = '0.0.0';}, + 'unfinished sender': value => {firstSender(value).sender_finished = false;}, + 'invented Python receipt': value => {firstSender(value).client_result = copy(firstSender(value).requests[0].response);}, + 'second run': value => {replacement(value).run_id = 'other-run';}, + 'second workflow callback': value => {replacement(value).workflow_invocations = 2;}, + 'activity callback': value => {replacement(value).activity_invocations = 1;}, + 'wrong checkpoint queue': value => {replacement(value).snapshot.task_queue = 'other';}, + 'wrong checkpoint namespace': value => {replacement(value).snapshot.namespace = 'other';}, + 'original not waiting': value => {value.sdk_signal_direction.phases[0].snapshot.status = 'completed';}, + 'first signal missing': value => {value.sdk_signal_direction.phases[2].snapshot.events.pop();}, + 'cold history truncated': value => {replacement(value).requests[1].response.task.history_events.pop();}, + 'wrong claimed run': value => {replacement(value).requests[1].response.task.run_id = 'other';}, + 'wrong claimed queue': value => {replacement(value).requests[1].response.task.task_queue = 'other';}, + 'registration refused': value => {replacement(value).requests[0].response.registered = false;}, + 'extra activity type': value => {replacement(value).requests[0].request.supported_activity_types = ['activity'];}, + 'completion refused': value => {completion(value).response.recorded = false;}, + 'stale lease': value => {completion(value).request.lease_owner = 'stale';}, + 'stale attempt': value => {completion(value).request.workflow_task_attempt = 2;}, + 'wrong task receipt': value => {completion(value).response.task_id = 'other';}, + 'cold replay reopens wait': value => {completion(value).request.commands[0].type = 'open_condition_wait';}, + 'unbound completion copy': value => {value.workflow_completions[2].response.recorded = false;}, + 'result frame replaced': value => {value.execution.output_envelope = {codec: 'avro', blob: 'other'};}, + 'int64 sender rounded': value => {firstSender(value).requests[0].typed_decoded_input.value[0].value.count.value = '9007199254740992';}, + 'int64 result rounded': value => {value.workflow_completions[2].decoded_commands[0].typed_decoded.result.value.count.value = '9007199254740992';}, + 'new start before cold replay': value => {replacement(value).requests.push(copy(value.sdk_signal_direction.phases[0].requests[1]));}, + 'start before registration': value => {value.sdk_signal_direction.phases[0].requests.reverse();}, + 'claim before accepted start': value => { + const requests = value.sdk_signal_direction.phases[0].requests; + [requests[1], requests[2]] = [requests[2], requests[1]]; + }, + 'signal refused': value => {firstSender(value).requests[0].status = 409;}, + 'created signal status': value => {firstSender(value).requests[0].status = 201;}, + 'wrong accepted run': value => {firstSender(value).requests[0].response.run_id = 'other';}, + 'run selector added': value => {firstSender(value).requests[0].request.run_id = value.run_id;}, + 'unbound delivery receipt': value => {value.signal_deliveries[0].response.command_id = 'other';}, + 'coalesced signal command': value => {value.sdk_signal_direction.phases[3].requests[0].response.command_id = firstSender(value).requests[0].response.command_id;}, + 'wrong next wait key': value => {value.sdk_signal_direction.phases[2].requests.at(-1).request.commands[0].condition_key = 'payload:0';}, + 'changed original wait fingerprint': value => {replacement(value).requests[1].response.task.history_events[2].payload.condition_definition_fingerprint = 'sha256:' + 'b'.repeat(64);}, +}; +for (const [name, mutate] of Object.entries(mutations)) test('refuse signal direction ' + name, () => { + const value = observed(fixture); mutate(value); + assert.throws(() => checkObservation(fixture, value, workflowId)); +}); +test('signal package guard refuses source and tuple substitutions', () => { + const value = observed(fixture); value.sdk_signal_direction.packages.python.archive_sha256 = 'a'.repeat(64); + assert.throws(() => checkSdkSignalDirectionPackages(fixture, value, artifacts)); + const altered = copy(artifacts); altered.server.source_commit = 'b'.repeat(40); + assert.throws(() => checkSdkSignalDirectionArtifacts(fixture, altered)); +}); + +for (const [name, mutate] of Object.entries({ + 'missing authored occurrence': value => {delete value.sdk_signal_direction.phases[0].requests.at(-1).request.commands[0].condition_wait_occurrence_id;}, + 'reused authored occurrence': value => {value.sdk_signal_direction.phases[2].requests.at(-1).request.commands[0].condition_wait_occurrence_id = 'rust:condition-wait:0';}, + 'changed satisfied occurrence': value => {value.events[11].payload.condition_wait_occurrence_id = 'rust:condition-wait:0';}, +})) test('refuse Rust signal ' + name, () => { + const rustFixture = fixtureFor('rust', 'php'), value = observed(rustFixture); + mutate(value); + assert.throws(() => checkObservation(rustFixture, value, workflowId)); +});