diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 2e8d9e7de..7a738dc73 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -402,6 +402,7 @@ jobs: -v "$PWD:/source:ro" -w /source "$image" \ scripts/conformance/server-parity/mysql-schema.php compare php_ref native_ref > parity-evidence/catalog-comparison.json - name: Compare three runtimes, recover killed native nodes and refuse published PHP takeover + id: prepare_runtime_comparison run: | set -euo pipefail image=$(jq -r '.server.images["linux/amd64"] | sub("^durableworkflow/server@"; "ghcr.io/durable-workflow/server@")' tests/Fixtures/ServerParityBaselines/php-2.5.14.json) @@ -627,7 +628,7 @@ jobs: - name: Compare separate configured role-token cohort run: bash scripts/conformance/server-parity/role-profile.sh mysql - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() }} + if: ${{ !cancelled() && steps.prepare_runtime_comparison.outcome == 'success' }} # The preceding process-loss case deliberately kills native-a. run: bash scripts/conformance/server-parity/corrected-marker-shared-profile.sh mysql http://parity-native-b:8080 - name: Retire corrected cohort resources @@ -839,6 +840,7 @@ jobs: -v "$PWD:/source:ro" -w /source "$image" \ scripts/conformance/server-parity/postgres-schema.php compare php_ref native_ref > parity-evidence/catalog-comparison.json - name: Compare three runtimes, recover killed native nodes and refuse published PHP takeover + id: prepare_runtime_comparison env: POSTGRES_IMAGE: ${{ matrix.image }} run: | @@ -1067,7 +1069,7 @@ jobs: POSTGRES_IMAGE: ${{ matrix.image }} run: bash scripts/conformance/server-parity/role-profile.sh pgsql - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() }} + if: ${{ !cancelled() && steps.prepare_runtime_comparison.outcome == 'success' }} env: POSTGRES_IMAGE: ${{ matrix.image }} # The preceding process-loss case deliberately kills native-a. @@ -1272,6 +1274,7 @@ jobs: cargo run --locked --offline --example codec-fixtures -- \ ../tests/Fixtures/ServerParity/Codec/v1.json /evidence/php-codec.json > parity-evidence/cross-decode.json - name: Execute and compare Rust, published PHP and embedded fixtures + id: prepare_runtime_comparison env: RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} run: | @@ -1491,7 +1494,7 @@ jobs: RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} run: bash scripts/conformance/server-parity/role-profile.sh sqlite - name: Compare corrected markers, ranges, clocks and activity expiry - if: ${{ !cancelled() }} + if: ${{ !cancelled() && steps.prepare_runtime_comparison.outcome == 'success' }} env: RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} run: bash scripts/conformance/server-parity/corrected-marker-shared-profile.sh sqlite http://parity-recovered:8080 diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index ba9173f45..8d580ee61 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -6,11 +6,16 @@ The unpublished Rust crate has an Avro foundation and an opt-in SQLite/PostgreSQ execution slice. Full runtime parity, migration, performance improvement and cutover remain unqualified. -The latest merged foundation is [#450](https://github.com/durable-workflow/server/pull/450), -at `e603a245e`, following #445's memo transaction preparation, #446's bounded -continuation and #448's SDK memo authoring. Each exact candidate tree passes -its own 13-job shared suite across six databases. Separate fresh-main -qualification remains pending; no result is carried across changed trees. +The delivered native foundation is [#450](https://github.com/durable-workflow/server/pull/450), +following memo transactions, activity/continuation replay and SDK memo authoring +in #445/#446/#448. Their exact fresh-main trees pass all 13 shared jobs across +six databases. #446's first failed main query remains failed and unexplained; +its unchanged second attempt reran all 13 jobs, rather than one database. +The consumer runner [#457](https://github.com/durable-workflow/server/pull/457) +is merged as `303ffc633` with its qualified candidate tree. All six candidate +execution logs have 45 passing records; separate fresh-main qualification +[38668891437](https://github.com/durable-workflow/server/actions/runs/38668891437) +remains pending. No result is carried across changed trees. The merged native development runtime enables memo upserts, while full external transport discovery stays unadvertised. Frozen published PHP, embedded and SDK tuples remain separate from current PHP release work. Representative timer, signal, @@ -20,45 +25,54 @@ failed; later qualifying runs are separate observations. ## Work order and current status -The active integration order is: - -- [#445](https://github.com/durable-workflow/server/pull/445): canonical memo - transaction foundation merged `bb199cb71`; its separate fresh-main 13-job - qualification passes in 38645794230, with every database log audited. - This preparation kept the memo-update gate disabled. -- [#446](https://github.com/durable-workflow/server/pull/446): cold continue-as-new - after a genuine activity, merged `dff9e846`. Its identical whole tree passes - all 13 candidate jobs after one unchanged MariaDB retry. The original hosted - runner loss remains failed without an established cause. Fresh main is pending. -- [#448](https://github.com/durable-workflow/server/pull/448): actual SDK memo - authoring and continuation inheritance, merged `284f5b822`. Its exact whole - tree passes all 13 candidate jobs and required source checks. Fresh main and - external-memo policy remain pending. -- [#450](https://github.com/durable-workflow/server/pull/450): verified local - external input, workflow/activity results and activity arguments. Installed - PHP/native originals, including a two-activity batch and cold parent, complete. - Two-reference atomic rollback/cold delivery controls pass. Its exact whole - tree is merged as `e603a245e` after all 13 candidate jobs and required source - checks pass. Fresh main is pending. Those regression jobs do not imply new - external-consumer execution on every database. One local 100 ms transfer control - failed; three isolated controls and an unchanged full repeat pass, with the - original cause still unknown. +The pending #455/#458 stack will be promoted through #458 after its combined +main candidate passes all required checks. This delivers the reviewed parent +in the same tested main tree. Earlier admitted matrices remain separate +observations and continue; strict branch protection is preserved. + - [#455](https://github.com/durable-workflow/server/pull/455): opt-in, - namespace/queue-scoped null-task poll wake. Its combined matrix remains - pending. An earlier component's SQLite PHP target failed after an activity + namespace/queue-scoped null-task poll hint without claim or lease authority. + The earlier component passes all 13 jobs in 38647001027. Current combined + qualification remains pending in #458. Both PHP query middleware additions + and combined OpenAPI revision 55 stay intact. An earlier component failed after an activity completion response when the Rust actor's registration DELETE returned lock-pressure 503; later refusal/repair phases were not reached. That original stays failed. Idle/query cost and matched performance remain open. -- [#457](https://github.com/durable-workflow/server/pull/457): extend the actual - installed external consumers to separate PHP/native databases in all six - configurations. Its first local SQLite recipe completes five originals per - target and cold reads with exact typed values. Other database execution, - final-head hosted qualification and promotion remain pending. This does not - qualify automatic policy, S3 or completion-upload authority. +- [#458](https://github.com/durable-workflow/server/pull/458): version-1 activity + result completion uploads, combined with the parent poll hints. It has 70 + passing library tests (18 explicit ignores), 113 SQLite HTTP cases and three + focused authority/race controls on MySQL 8.4 and PostgreSQL 17. Actual 36 + published-PHP/native originals and strict readers pass across those three + local databases with genuine Unicode ownership, original deadlines, exact + payload hashes and cold parent completion. Full current-main/all-six + qualification and promotion remain pending. Initial Unicode, MySQL identity, + JSON member-order and 100ms control failures stay recorded in the PR. + The complete combined `3a268200` run `38680689284` includes qualified PHP + `ed803519`, passes twelve jobs and fails PostgreSQL 17 at the independent + fresh upload still using the preceding stalled request's 100 ms test policy. + All six small execution summaries were independently read: five cells pass + 113 execution tests; PostgreSQL 17 has 112 passes and that one failure. + That original remains failed. + The next fixture revision keeps the stalled deadline and shared semaphore, + then uses normal policy for the separate fresh upload. Recovered-target + checks will require their actual preparation step to succeed, preventing + dependent absence from looking like an execution failure. The next combined + source also integrates separately qualified PHP empty-command-contract + correction `35cd3642`; it needs its own full qualification. + No workflow/query/prepared + context, pressure-bypass or full transport capability is advertised. +- [#460](https://github.com/durable-workflow/server/pull/460): read-only upgrade + source inventory. Four actual published PHP 2.8.0 profiles pass locally, with + eight original pending runs, twelve refusals and unchanged source state. + WAL SQLite refuses before opening; all backup/quiescence/representation/ + takeover flags remain false. Combined hosted qualification and actual + conversion/backup/continuation gates remain ahead. Full transport discovery stays unadvertised. Complete batch workflow semantics, memo/query reference ingress, upload allowances, automatic result policy, S3, cleanup/backup/restart and full database/consumer qualification remain open. +Current PHP 2.8 query-authority and activity-result-inspection parity also +require separate native qualification. Broader runtime/operational parity, safe sequential PHP database takeover, matched performance measurements and release/cutover preparation remain #325 requirements. No native release or deployment is qualified. @@ -317,8 +331,17 @@ exact outputs and zero task repairs. They do not qualify a performance gain: PHP has two idle API timeouts and a forced scheduler shutdown, while native workflow tails vary. The original idle reader also rejects protocol-permitted `timeout` replies. Separate reader and scheduler controls pass, with original -failures preserved. Next diagnose native tails with a bounded published-worker -control, then repeat the corrected matched profile before a gain claim. +failures preserved. Separate installed-worker controls identify a native poll +wake gap: published PHP interrupts an opted-in activity/query wait when new +workflow work arrives, while the native baseline consumes the remaining waits. +This does not establish the cause of every earlier tail. The +[scoped correction in #455](../scripts/conformance/server-parity/CROSS-KIND-POLL-WAKE.md) +keeps task/lease authority in ordinary claims. Its five focused HTTP test groups +pass locally on SQLite, PostgreSQL 17 and MySQL 8.4; the unchanged native baseline +fails all three positive groups. Installed-worker activity/query controls also +pass on PHP and native SQLite with the original workflow/run identities. +Six-cell qualification of the new combined tree remains required before +repeating the corrected matched profile or claiming a gain. The #444 original MySQL 8.4 matrix on `95cd4efa` exposed a native activity clock disagreement: a retry was recorded 101 ms before its durable @@ -2880,3 +2903,11 @@ final result; native keeps those inline under its unadvertised automatic policy. Both containers stop with 137 and no OOM. All local databases, provider objects, SDK dependencies, compiler output and task resources are removed after this audit. Hosted all-database external semantics remain pending. + +The upload/poll-hint candidate now incorporates exact qualified PHP source +`1f994962` and its `applyStickyCacheClaim` primary-key correction, including +the current PHP release metadata and numeric declaration normalization. Its +WorkerController bytes match the qualified PHP handoff. This combined tree +requires fresh normal and six-database qualification. The preceding `3409a776` +component pass, its MySQL-only retry, and all failed originals remain their +separate records; none qualifies this integration or a native release. diff --git a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml index 8e3a60aa3..b10e932ac 100644 --- a/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml +++ b/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: durable-workflow.v2.worker-protocol-api - version: "54" + version: "55" summary: Durable Workflow worker-plane HTTP+JSON API description: > Normative OpenAPI specification for the worker-plane HTTP+JSON API: @@ -23,6 +23,14 @@ x-durable-workflow-catalog-entry: worker_protocol_api x-durable-workflow-catalog-schema: durable-workflow.v2.platform-protocol-specs.catalog x-durable-workflow-catalog-version: 16 x-durable-workflow-evolution-rule: additive_minor_breaking_major +x-durable-workflow-cross-kind-poll-wake: + opt_in_worker_capability: cross_kind_poll_wake + scope: [namespace, task_queue] + reply: { task: null, poll_status: task_queue_changed } + authority: hint_only + allocates_task_or_lease: false + changes_acceptance_or_deadline: false + non_opted_workers: preserve_existing_behavior x-durable-workflow-cancellation-scope-admission: minimum_protocol_version: "1.20" status: unfrozen_candidate_membership_only @@ -1965,7 +1973,15 @@ components: required: [task] properties: task: { type: "null" } - poll_status: { type: string, enum: [timeout, empty] } + poll_status: + type: string + enum: [timeout, empty, task_queue_changed] + description: > + task_queue_changed is an optional hint for workers registered + with cross_kind_poll_wake. It interrupts an empty wait after + work is published in that namespace and task queue so the worker + can sweep its task kinds again. It grants no task, lease or + acceptance authority and does not renew an existing deadline. DrainingPoll: allOf: - $ref: "#/components/schemas/WorkerEnvelope" diff --git a/rust/README.md b/rust/README.md index 512e8184c..b2619e628 100644 --- a/rust/README.md +++ b/rust/README.md @@ -283,8 +283,9 @@ The development external-payload routes upload/fetch verified encoded bytes at `/api/external-payloads/v1`, with authenticated namespace binding, local policy, object/byte quota admission and an independent request size/timeout bound. `PUT /api/namespaces/{namespace}/external-storage` is admin-only and currently -supports the local driver. Completion-upload allowance headers refuse until -their original task/lease authority is implemented. Full transport discovery +supports the local driver. Version-1 activity-result completion uploads require +the original task/attempt/owner and preserve a bounded allowance without renewing +the lease. Workflow/query/prepared completion contexts remain refused. Full transport discovery stays unadvertised pending real workflow resolution/retention, cleanup/backup, restart, all database families and S3 qualification. The actual byte probe is `scripts/conformance/server-parity/external-payload-http.py`; run it only against @@ -300,8 +301,10 @@ result-upload/public task-completion recipe and separate cold SDK reader are in `scripts/conformance/server-parity/external-result-client.php`. Its `activity-author` and `activity-reader` phases use explicit activity-result HTTP completion and a separate installed SDK workflow-history reader. These recipes -do not establish automatic SDK upload discovery. Memo/query ingress, -completion-upload allowances and automatic external result policy remain +do not establish automatic SDK upload discovery. The separate raw HTTP recipe +`scripts/conformance/server-parity/activity-completion-upload-probe.php` covers +the activity completion-upload context, including real UTF-8 JSON. Memo/query +ingress, broader completion contexts and automatic external result policy remain unqualified. Apache's schema-aware validation patch is recorded in `vendor/apache-avro/PATCHES.md`. diff --git a/rust/src/runtime/activity_completion_uploads.rs b/rust/src/runtime/activity_completion_uploads.rs new file mode 100644 index 000000000..39d56486b --- /dev/null +++ b/rust/src/runtime/activity_completion_uploads.rs @@ -0,0 +1,441 @@ +//! The published v1 activity-result upload allowance grants no task completion. +use super::{ + Result, RuntimeError, + backend::Backend, + refuse, + store::{Store, now}, +}; +use crate::external_payload::Reference; +use axum::http::StatusCode; +use chrono::{DateTime, Utc}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Transaction, Type}; + +#[derive(Clone)] +pub(crate) struct Context { + value: Value, + task: String, + attempt: String, + owner: String, +} + +fn invalid() -> RuntimeError { + refuse( + StatusCode::UNPROCESSABLE_ENTITY, + "external_payload_completion_invalid", + ) +} +fn rejected() -> RuntimeError { + refuse( + StatusCode::CONFLICT, + "external_payload_completion_lease_rejected", + ) +} +fn unavailable() -> RuntimeError { + refuse( + StatusCode::SERVICE_UNAVAILABLE, + "external_payload_unavailable", + ) +} +fn lease_error(error: RuntimeError) -> RuntimeError { + match error { + RuntimeError::Refused { .. } => rejected(), + error => error, + } +} +fn identifier(value: &Value) -> Result { + value + .as_str() + .filter(|value| { + !value.is_empty() + && value.len() <= 255 + && !value.chars().any(|c| c <= '\u{1f}' || c == '\u{7f}') + }) + .map(str::to_owned) + .ok_or_else(invalid) +} + +// PHP's default json_encode escapes slashes and UTF-16 code units. These hashes +// must also match retained PHP budget identities during a later database takeover. +fn identity(value: &Value) -> String { + use std::fmt::Write; + let mut encoded = String::new(); + for c in serde_json::to_string(value).expect("JSON value").chars() { + if c == '/' { + encoded.push_str("\\/"); + } else if c.is_ascii() { + encoded.push(c); + } else { + for unit in c.encode_utf16(&mut [0; 2]) { + write!(encoded, "\\u{unit:04x}").expect("String write"); + } + } + } + Sha256::digest(encoded.as_bytes()) + .iter() + .map(|b| format!("{b:02x}")) + .collect() +} + +impl Context { + pub(super) fn parse_bytes(header: &[u8]) -> Result { + if header.len() > 4096 { + return Err(invalid()); + } + Self::parse(std::str::from_utf8(header).map_err(|_| invalid())?) + } + pub(super) fn parse(header: &str) -> Result { + if header.len() > 4096 { + return Err(invalid()); + } + let value: Value = serde_json::from_str(header).map_err(|_| invalid())?; + if value["schema"] == "durable-workflow.v2.payload-completion-context.v2" { + return Err(refuse( + StatusCode::UNPROCESSABLE_ENTITY, + "development_payload_completion_upload_unqualified", + )); + } + let fields = [ + "schema", + "kind", + "task_id", + "attempt", + "lease_owner", + "operation", + "slot", + ]; + if value["schema"] != "durable-workflow.v2.payload-completion-context.v1" + || value.as_object().is_none_or(|map| { + map.len() != fields.len() || fields.iter().any(|key| !map.contains_key(*key)) + }) + { + return Err(invalid()); + } + if value["kind"] != "activity" || value["operation"] != "complete" { + return Err(refuse( + StatusCode::UNPROCESSABLE_ENTITY, + "development_payload_completion_upload_unqualified", + )); + } + if value["slot"] != json!(["result"]) { + return Err(invalid()); + } + Ok(Self { + task: identifier(&value["task_id"])?, + attempt: identifier(&value["attempt"])?, + owner: identifier(&value["lease_owner"])?, + value, + }) + } + fn scope(&self, namespace: &str) -> String { + identity(&json!([ + namespace, + "activity", + self.task, + self.attempt, + self.owner + ])) + } + fn slot(&self) -> String { + identity(&json!(["complete", ["result"]])) + } +} + +impl Store +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> &'q str: Encode<'q, DB> + Type, + for<'q> Option<&'q str>: Encode<'q, DB>, + for<'q> Option: Encode<'q, DB>, + for<'q> String: Encode<'q, DB> + Type, + for<'q> i64: Encode<'q, DB> + Type, + for<'r> i64: Decode<'r, DB>, + usize: ColumnIndex, +{ + pub(super) async fn completion_upload_lease( + &self, + tx: &mut Transaction<'_, DB>, + context: &Context, + ) -> Result> { + let task = self + .task_row(tx, &context.task) + .await + .map_err(lease_error)?; + // A case-folding database lookup must not create another allowance for + // an alias of the same canonical task or attempt. + if DB::string(&task, "task_type")? != "activity" || DB::string(&task, "id")? != context.task + { + return Err(rejected()); + } + Self::fence(&task, &json!({"lease_owner":context.owner})).map_err(lease_error)?; + let attempt = + Self::query("SELECT * FROM activity_attempts WHERE id=$1 AND workflow_task_id=$2") + .bind(&context.attempt) + .bind(&context.task) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(rejected)?; + let run_id = DB::string(&task, "workflow_run_id")?; + Self::active_run(tx, &run_id).await.map_err(lease_error)?; + let activity = + Self::query("SELECT * FROM activity_executions WHERE id=$1 AND workflow_run_id=$2") + .bind(DB::string(&attempt, "activity_execution_id")?) + .bind(&run_id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(rejected)?; + if DB::string(&attempt, "status")? != "running" + || DB::string(&attempt, "id")? != context.attempt + || DB::number(&attempt, "attempt_number")? != DB::number(&task, "attempt_count")? + || DB::string(&attempt, "lease_owner")? != context.owner + || DB::optional_string(&activity, "current_attempt_id")?.as_deref() + != Some(context.attempt.as_str()) + || DB::number(&activity, "attempt_count")? != DB::number(&attempt, "attempt_number")? + || DB::string(&activity, "status")? != "running" + { + return Err(rejected()); + } + super::activity_deadlines::fence::(&activity).map_err(lease_error)?; + DB::optional_instant(&task, "lease_expires_at")?.ok_or_else(rejected) + } + + async fn completion_upload_reference( + &self, + tx: &mut Transaction<'_, DB>, + context: &Context, + expected: &Reference, + ) -> Result> { + let scope = context.scope(&self.namespace); + let Some(budget) = Self::query( + "SELECT slots FROM runtime_payload_completion_budgets WHERE id=$1 AND namespace=$2", + ) + .bind(&scope) + .bind(self.namespace.as_ref()) + .fetch_optional(&mut **tx) + .await? + else { + return Ok(None); + }; + let slots = DB::document_row(&budget, "slots")?; + let slot = &slots[context.slot()]; + if slot["sha256"] != expected.sha256 || slot["size_bytes"] != expected.size_bytes { + return Ok(None); + } + let Ok(reference) = Reference::validate(&slot["reference"]) else { + return Ok(None); + }; + let Some(row) = + Self::query("SELECT * FROM runtime_external_payloads WHERE id=$1 AND namespace=$2") + .bind(&reference.reference_id) + .bind(self.namespace.as_ref()) + .fetch_optional(&mut **tx) + .await? + else { + return Ok(None); + }; + if DB::string(&row, "codec")? != "avro" + || DB::string(&row, "sha256")? != expected.sha256 + || DB::number(&row, "size_bytes")? != expected.size_bytes + || DB::string(&row, "upload_status")? != "ready" + || DB::optional_instant(&row, "retained_at")?.is_none() + && DB::optional_instant(&row, "expires_at")?.is_none_or(|expires| expires <= now()) + { + return Ok(None); + } + Ok(Some(reference)) + } + + pub(super) async fn authorize_completion_upload( + &self, + context: &Context, + expected: &Reference, + ) -> Result> { + let mut tx = self.begin().await?; + let existing = self + .completion_upload_reference(&mut tx, context, expected) + .await?; + if existing.is_none() { + self.completion_upload_lease(&mut tx, context).await?; + } + tx.commit().await?; + Ok(existing) + } + + pub(super) async fn upload_activity_completion( + &self, + local: crate::external_payload::LocalStorage, + expected: Reference, + snapshot: crate::external_payload::Snapshot, + expires: DateTime, + quota: super::external_config::Quota, + context: Context, + max_bytes: u64, + retention: i64, + ) -> Result { + let scope = context.scope(&self.namespace); + let slot = context.slot(); + let mut tx = self.begin().await?; + Self::payload_object_lock(&mut tx, &scope).await?; + if let Some(reference) = self + .completion_upload_reference(&mut tx, &context, &expected) + .await? + { + return Ok(reference); + } + let lease = self.completion_upload_lease(&mut tx, &context).await?; + let expiry = chrono::Duration::try_seconds(retention) + .and_then(|duration| lease.checked_add_signed(duration)) + .ok_or_else(unavailable)?; + let budget = Self::query("SELECT slots,objects FROM runtime_payload_completion_budgets WHERE id=$1 AND namespace=$2") + .bind(&scope).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?; + let (mut slots, mut objects) = match &budget { + Some(row) => ( + DB::document_row(row, "slots")?, + DB::document_row(row, "objects")?, + ), + None => (json!({}), json!({})), + }; + let slots_map = slots.as_object_mut().ok_or_else(unavailable)?; + let objects_map = objects.as_object_mut().ok_or_else(unavailable)?; + let recorded = slots_map.get(&slot); + if recorded.is_some_and(|value| { + value["sha256"] != expected.sha256 || value["size_bytes"] != expected.size_bytes + }) { + return Err(refuse( + StatusCode::CONFLICT, + "external_payload_completion_conflict", + )); + } + let total = objects_map + .values() + .try_fold(0_u64, |sum, size| sum.checked_add(size.as_u64()?)) + .ok_or_else(unavailable)?; + if recorded.is_none() && slots_map.len() >= 128 + || !objects_map.contains_key(&expected.sha256) + && expected.size_bytes as u64 > max_bytes.saturating_sub(total) + { + return Err(refuse( + StatusCode::TOO_MANY_REQUESTS, + "external_payload_completion_budget_exhausted", + )); + } + slots_map + .entry(slot.clone()) + .or_insert_with(|| json!({"sha256":expected.sha256,"size_bytes":expected.size_bytes})); + objects_map.insert(expected.sha256.clone(), json!(expected.size_bytes)); + let at = now(); + if budget.is_none() { + Self::query("INSERT INTO runtime_payload_completion_budgets(id,namespace,context,slots,objects,expires_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)") + .bind(&scope).bind(self.namespace.as_ref()).bind(DB::document(&context.value)).bind(DB::document(&slots)) + .bind(DB::document(&objects)).bind(DB::bind_time(expiry)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).execute(&mut *tx).await?; + } else { + Self::query("UPDATE runtime_payload_completion_budgets SET slots=$1,objects=$2,expires_at=$3,updated_at=$4 WHERE id=$5 AND namespace=$6") + .bind(DB::document(&slots)).bind(DB::document(&objects)).bind(DB::bind_time(expiry)).bind(DB::bind_time(at)) + .bind(&scope).bind(self.namespace.as_ref()).execute(&mut *tx).await?; + } + tx.commit().await?; + let reference = self + .upload_external_local_guarded( + local, + expected, + snapshot, + expires, + quota, + Some(&context), + ) + .await?; + let mut tx = self.begin().await?; + Self::payload_object_lock(&mut tx, &scope).await?; + let row = Self::query( + "SELECT slots FROM runtime_payload_completion_budgets WHERE id=$1 AND namespace=$2", + ) + .bind(&scope) + .bind(self.namespace.as_ref()) + .fetch_one(&mut *tx) + .await?; + let mut slots = DB::document_row(&row, "slots")?; + slots[&slot]["reference"] = json!(reference); + Self::query("UPDATE runtime_payload_completion_budgets SET slots=$1,updated_at=$2 WHERE id=$3 AND namespace=$4") + .bind(DB::document(&slots)).bind(DB::bind_time(now())).bind(&scope).bind(self.namespace.as_ref()).execute(&mut *tx).await?; + tx.commit().await?; + Ok(reference) + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn malformed_contexts_refuse_without_becoming_upload_authority() { + let valid = json!({"schema":"durable-workflow.v2.payload-completion-context.v1","kind":"activity", + "task_id":"task","attempt":"attempt","lease_owner":"owner","operation":"complete","slot":["result"]}); + assert!(Context::parse(&valid.to_string()).is_ok()); + for (field, value) in [ + ("attempt", json!(1)), + ("lease_owner", json!("owner\n")), + ("task_id", json!("")), + ("slot", json!(["commands", 0, "result"])), + ] { + let mut bad = valid.clone(); + bad[field] = value; + assert!( + matches!( + Context::parse(&bad.to_string()), + Err(RuntimeError::Refused { + reason: "external_payload_completion_invalid", + .. + }) + ), + "{bad}" + ); + } + let mut extra = valid; + extra["new_authority"] = json!(true); + assert!(Context::parse(&extra.to_string()).is_err()); + assert!(Context::parse(&" ".repeat(4097)).is_err()); + } + #[test] + fn scope_matches_actual_published_php_ascii_slash_unicode_and_surrogate_hashes() { + assert_eq!( + identity(&json!(["default", "activity", "task", "attempt", "owner"])), + "eb0609d789d53c946f889b345ed8231ec6e217431628f1c5d59ccb0d24867427" + ); + assert_eq!( + identity(&json!([ + "default", + "activity", + "task", + "attempt", + "owner/位馃榾" + ])), + "3ed044a74929ada4c313de51b22715f4e974a0a12c7356330500bc736df66641" + ); + assert_eq!( + identity(&json!(["complete", ["result"]])), + "be37893a3312755fa3826ca9282c3bd053521dc77235d9024c4c58de6e49dcd0" + ); + } + #[test] + fn completion_header_accepts_utf8_and_preserves_escaped_identity() { + let raw = json!({"schema":"durable-workflow.v2.payload-completion-context.v1", + "kind":"activity", "task_id":"task", "attempt":"attempt", + "lease_owner":"owner/位馃榾", "operation":"complete", "slot":["result"]}) + .to_string(); + let escaped = raw.replace('位', "\\u03bb").replace('馃榾', "\\ud83d\\ude00"); + let raw = Context::parse_bytes(raw.as_bytes()).unwrap(); + let escaped = Context::parse_bytes(escaped.as_bytes()).unwrap(); + assert_eq!(raw.owner, "owner/位馃榾"); + assert_eq!(raw.scope("default"), escaped.scope("default")); + for malformed in [vec![0xff], vec![b' '; 4097]] { + assert!(matches!( + Context::parse_bytes(&malformed), + Err(RuntimeError::Refused { + reason: "external_payload_completion_invalid", + .. + }) + )); + } + } +} diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 2db9ecd9c..cfee23605 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -15,6 +15,8 @@ pub(super) trait Backend: Database { const VISIBILITY_PAGE: &'static str = super::visibility::VISIBILITY_PAGE; const WORKER_REGISTRATION_SQL: &'static str = super::store::WORKER_REGISTRATION_SQL; const TASK_CANDIDATES_SQL: &'static str = super::store::TASK_CANDIDATES_SQL; + const POLL_WAKE_TASKS_SQL: &'static str = super::cross_kind_poll::TASKS_SQL; + const POLL_WAKE_QUERIES_SQL: &'static str = "SELECT \"key\" AS id FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration>$1 AND CAST(value AS jsonb)->>'namespace'=$2 AND CAST(value AS jsonb)->>'task_queue'=$3 AND CAST(CAST(value AS jsonb)->>'deadline' AS BIGINT)>$4 AND (CAST(value AS jsonb)->>'status'='pending' OR (CAST(value AS jsonb)->>'status'='leased' AND CAST(CAST(value AS jsonb)->>'lease_until' AS BIGINT)<=$5)) ORDER BY \"key\" LIMIT 65"; const POLL_RECEIPT_CLEANUP_SQL: &'static str = super::store::POLL_RECEIPT_CLEANUP_SQL; const TIMER_TASK_SQL: &'static str; const SELECTION_RESOLUTION_SQL: &'static str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND payload->>'selection_group_id'=$2 ORDER BY sequence LIMIT 1"; @@ -84,6 +86,25 @@ fn sqlite_instant(value: String) -> Result> { } impl Backend for Sqlite { + const POLL_WAKE_QUERIES_SQL: &'static str = "SELECT \"key\" AS id FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration>$1 AND json_extract(value,'$.namespace')=$2 AND json_extract(value,'$.task_queue')=$3 AND json_extract(value,'$.deadline')>$4 AND (json_extract(value,'$.status')='pending' OR (json_extract(value,'$.status')='leased' AND json_extract(value,'$.lease_until')<=$5)) ORDER BY \"key\" LIMIT 65"; + const POLL_WAKE_TASKS_SQL: &'static str = r#" +WITH utc_tasks AS ( + SELECT t.id,t.namespace,t.queue,t.workflow_run_id,t.status,t.available_at, + replace(replace(replace(t.available_at,'T',' '),'Z',''),'+00:00','') AS available_utc, + replace(replace(replace(t.lease_expires_at,'T',' '),'Z',''),'+00:00','') AS lease_utc + FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') +), ordered_tasks AS ( + SELECT *, + rtrim(rtrim(CASE WHEN length(available_utc)=19 THEN available_utc||'.0' ELSE available_utc END,'0'),'.') AS available_order, + rtrim(rtrim(CASE WHEN length(lease_utc)=19 THEN lease_utc||'.0' ELSE lease_utc END,'0'),'.') AS lease_order + FROM utc_tasks +) +SELECT t.id,t.namespace,t.queue FROM ordered_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id +WHERE r.status IN ('pending','running','waiting') + AND (t.status='ready' OR (t.status='leased' AND t.lease_order<=rtrim(rtrim($4,'0'),'.'))) + AND (t.available_at IS NULL OR t.available_order<=rtrim(rtrim($5,'0'),'.')) +ORDER BY t.available_order,t.id LIMIT 100 +"#; const QUERY_CACHE_PENDING_FOR_RUN: &'static str = "SELECT COUNT(*) FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND expiration>$2 AND json_extract(value,'$.run_id')=$1 AND json_extract(value,'$.status') IN ('pending','leased')"; const SELECTION_RESOLUTION_SQL: &'static str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND json_extract(payload,'$.selection_group_id')=$2 ORDER BY sequence LIMIT 1"; const REPAIR_COUNTER_MAX: i64 = i64::MAX; @@ -287,6 +308,7 @@ impl Backend for Postgres { } impl Backend for MySql { + const POLL_WAKE_QUERIES_SQL: &'static str = "SELECT `key` AS id FROM dw_query_cache WHERE `key` LIKE 'dw-native-query:%' AND expiration>? AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.namespace')) AS BINARY)=CAST(? AS BINARY) AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.task_queue')) AS BINARY)=CAST(? AS BINARY) AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.deadline')) AS SIGNED)>? AND (JSON_UNQUOTE(JSON_EXTRACT(value,'$.status'))='pending' OR (JSON_UNQUOTE(JSON_EXTRACT(value,'$.status'))='leased' AND CAST(JSON_UNQUOTE(JSON_EXTRACT(value,'$.lease_until')) AS SIGNED)<=?)) ORDER BY `key` LIMIT 65"; const EXTERNAL_PAYLOAD_BYTES: &'static str = "SELECT CAST(COALESCE(SUM(size_bytes),0) AS SIGNED) FROM runtime_external_payloads WHERE namespace=$1"; const MEMO_UPSERT: &'static str = "INSERT INTO workflow_memos(workflow_run_id,workflow_instance_id,`key`,value,upserted_at_sequence,inherited_from_parent,created_at,updated_at,portable_value,portable_value_sequence) VALUES ($1,$2,$3,$4,$5,false,$6,$7,$8,$9) ON DUPLICATE KEY UPDATE value=VALUES(value),upserted_at_sequence=VALUES(upserted_at_sequence),inherited_from_parent=false,updated_at=VALUES(updated_at),portable_value=VALUES(portable_value),portable_value_sequence=VALUES(portable_value_sequence)"; const MEMO_DELETE: &'static str = @@ -324,6 +346,7 @@ impl Backend for MySql { || sql == Self::RANGE_REPAIR_TASK_SQL || sql == Self::SELECTION_RESOLUTION_SQL || sql == Self::ACTIVITY_OPEN_TASK_SQL + || sql == Self::POLL_WAKE_QUERIES_SQL || sql == Self::QUERY_CACHE_PENDING_FOR_RUN { return Cow::Borrowed(sql); diff --git a/rust/src/runtime/cross_kind_poll.rs b/rust/src/runtime/cross_kind_poll.rs new file mode 100644 index 000000000..f780b5c5b --- /dev/null +++ b/rust/src/runtime/cross_kind_poll.rs @@ -0,0 +1,115 @@ +//! Optional, scoped hints. Readiness observation never claims another task kind. +use super::{Result, backend::Backend, store::Store, text}; +use serde_json::Value; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; + +pub(super) struct PollWakeSnapshot { + registration: String, + ready: Vec, +} + +pub(super) const TASKS_SQL: &str = "SELECT t.id,t.namespace,t.queue FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100"; + +impl Store +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> &'q str: Encode<'q, DB> + Type, + for<'q> Option<&'q str>: Encode<'q, DB>, + for<'q> Option: Encode<'q, DB>, + for<'q> String: Encode<'q, DB> + Type, + for<'q> i64: Encode<'q, DB> + Type, + for<'r> i64: Decode<'r, DB>, + usize: ColumnIndex, +{ + pub(super) async fn poll_wake_snapshot( + &self, + body: &Value, + kind: &str, + ) -> Result> { + let worker_id = text(body, "worker_id")?; + let queue = text(body, "task_queue")?; + let worker = Self::query("SELECT w.capabilities,w.namespace,w.task_queue,w.worker_id,i.token FROM workflow_worker_registrations w JOIN dw_worker_registration_incarnations i ON i.namespace=w.namespace AND i.worker_id=w.worker_id AND i.status='active' WHERE w.namespace=$1 AND w.worker_id=$2 AND w.task_queue=$3 AND w.status='active'") + .bind(self.namespace.as_ref()).bind(text(body,"worker_id")?).bind(text(body,"task_queue")?) + .fetch_optional(&self.pool).await?; + let Some(worker) = worker else { + return Ok(None); + }; + if DB::string(&worker, "namespace")? != self.namespace.as_ref() + || DB::string(&worker, "task_queue")? != queue + || DB::string(&worker, "worker_id")? != worker_id + { + return Ok(None); + } + if !DB::optional_document_row(&worker, "capabilities")?.is_some_and(|caps| { + caps.as_array() + .is_some_and(|caps| caps.iter().any(|cap| cap == "cross_kind_poll_wake")) + }) { + return Ok(None); + } + // Only bounded routing metadata leaves the database, never task payloads or query + // history snapshots. The same reads observe independent server nodes. + let at = super::store::now(); + let rows = Self::query(DB::POLL_WAKE_TASKS_SQL) + .bind(self.namespace.as_ref()) + .bind(text(body, "task_queue")?) + .bind(kind) + .bind(DB::bind_time(at)) + .bind(DB::bind_time(at)) + .fetch_all(&self.pool) + .await?; + let mut ready = Vec::new(); + for row in rows { + // Keep the indexed SQL scope filter, then compare exact bytes: + // the preserved MySQL collation folds case and accents. + if DB::string(&row, "namespace")? == self.namespace.as_ref() + && DB::string(&row, "queue")? == queue + { + ready.push(format!("task:{}", DB::string(&row, "id")?)); + } + } + if kind != "query" { + ready.extend( + Self::query(DB::POLL_WAKE_QUERIES_SQL) + .bind(at.timestamp()) + .bind(self.namespace.as_ref()) + .bind(text(body, "task_queue")?) + .bind(at.timestamp()) + .bind(at.timestamp()) + .fetch_all(&self.pool) + .await? + .iter() + .map(|row| DB::string(row, "id").map(|id| format!("query:{id}"))) + .collect::>>()?, + ); + } + Ok(Some(PollWakeSnapshot { + registration: DB::string(&worker, "token")?, + ready, + })) + } + + pub(super) async fn poll_queue_changed( + &self, + body: &Value, + kind: &str, + previous: &mut Option, + ) -> Result { + let Some(before) = previous else { + return Ok(false); + }; + let Some(current) = self.poll_wake_snapshot(body, kind).await? else { + *previous = None; + return Ok(false); + }; + // A replacement registration cannot lend its opt-in to a held poll. + // The original claim path still owns all registration/lease fences. + if current.registration != before.registration { + *previous = None; + return Ok(false); + } + let changed = current.ready.iter().any(|id| !before.ready.contains(id)); + *previous = Some(current); + Ok(changed) + } +} diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 132d165c8..d9ab2be00 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -272,6 +272,24 @@ impl Runtime { expected: &crate::external_payload::Reference, ) -> Result ); + delegate!( + authorize_completion_upload( + context: &super::activity_completion_uploads::Context, + expected: &crate::external_payload::Reference, + ) -> Result> + ); + delegate!( + upload_activity_completion( + local: crate::external_payload::LocalStorage, + expected: crate::external_payload::Reference, + snapshot: crate::external_payload::Snapshot, + expires: chrono::DateTime, + quota: super::external_config::Quota, + context: super::activity_completion_uploads::Context, + max_bytes: u64, + retention: i64, + ) -> Result + ); delegate!(mark_external_fetched(expected: &crate::external_payload::Reference) -> Result<()>); delegate!(database_live() -> bool); pub(crate) async fn start(&self, body: Value) -> Result { diff --git a/rust/src/runtime/external_config.rs b/rust/src/runtime/external_config.rs index 0002f582c..e4992b199 100644 --- a/rust/src/runtime/external_config.rs +++ b/rust/src/runtime/external_config.rs @@ -7,6 +7,7 @@ use tokio::sync::Semaphore; #[derive(Clone)] pub struct ExternalPayloadConfig { pub max_bytes: u64, + pub completion_max_bytes: Option, pub timeout: Duration, pub abandoned_seconds: i64, pub max_bytes_per_namespace: Value, @@ -27,6 +28,7 @@ impl Default for ExternalPayloadConfig { fn default() -> Self { Self { max_bytes: 64 * 1024 * 1024, + completion_max_bytes: None, timeout: Duration::from_secs(30), abandoned_seconds: 3600, max_bytes_per_namespace: Value::Null, @@ -119,12 +121,15 @@ fn member<'a>(value: &'a Value, key: &str) -> Option<&'a Value> { impl ExternalPayloadConfig { pub(super) fn from_env() -> Result { + let completion = process_value("COMPLETION_MAX_BYTES", Value::Null)?; let overrides = process_value("NAMESPACE_OVERRIDES", Value::String("{}".into()))?; let overrides = overrides .as_str() .and_then(|value| serde_json::from_str(value).ok()) .unwrap_or(Value::Null); Ok(Self { + completion_max_bytes: (!completion.is_null()) + .then(|| integer(&completion).max(1) as u64), max_bytes: integer(&process_value( "MAX_BYTES", serde_json::json!(64 * 1024 * 1024), diff --git a/rust/src/runtime/external_registry.rs b/rust/src/runtime/external_registry.rs index 3c163bece..9f591a167 100644 --- a/rust/src/runtime/external_registry.rs +++ b/rust/src/runtime/external_registry.rs @@ -120,6 +120,19 @@ where snapshot: crate::external_payload::Snapshot, expires: DateTime, quota: Quota, + ) -> Result { + self.upload_external_local_guarded(local, expected, snapshot, expires, quota, None) + .await + } + + pub(super) async fn upload_external_local_guarded( + &self, + local: crate::external_payload::LocalStorage, + expected: Reference, + snapshot: crate::external_payload::Snapshot, + expires: DateTime, + quota: Quota, + completion: Option<&super::activity_completion_uploads::Context>, ) -> Result { let uri = local .uri_for(&expected) @@ -136,10 +149,16 @@ where .fetch_one(&mut *tx).await? != 1 { return Err(refuse(StatusCode::SERVICE_UNAVAILABLE, "external_payload_unavailable")); } + if let Some(context) = completion { + self.completion_upload_lease(&mut tx, context).await?; + } local .put(reference.clone(), snapshot) .await .map_err(super::external_transport::snapshot_error)?; + if let Some(context) = completion { + self.completion_upload_lease(&mut tx, context).await?; + } self.promote_external_upload(&mut tx, &reference, &uri) .await?; tx.commit().await?; diff --git a/rust/src/runtime/external_transport.rs b/rust/src/runtime/external_transport.rs index 358b1836a..db6f910c9 100644 --- a/rust/src/runtime/external_transport.rs +++ b/rust/src/runtime/external_transport.rs @@ -32,6 +32,8 @@ pub(super) fn error_body(status: StatusCode, reason: &str) -> Value { "reason":reason,"message":reason,"retryable":retryable,"status":status.as_u16()}); if reason.starts_with("external_payload_namespace_") { body["retry_after_seconds"] = json!(60); + } else if reason == "external_payload_completion_budget_exhausted" { + body["retry_after_seconds"] = json!(5); } body } @@ -50,6 +52,10 @@ fn response_error(error: RuntimeError) -> Response { response .headers_mut() .insert("retry-after", "60".parse().expect("static header")); + } else if reason == "external_payload_completion_budget_exhausted" { + response + .headers_mut() + .insert("retry-after", "5".parse().expect("static header")); } response } @@ -217,15 +223,11 @@ async fn upload_inner(runtime: Runtime, request: Request, deadline: Instant) -> "external_payload_unsupported", )); } - if request + let context = request .headers() - .contains_key("x-durable-workflow-payload-completion") - { - return Err(refuse( - StatusCode::UNPROCESSABLE_ENTITY, - "development_payload_completion_upload_unqualified", - )); - } + .get("x-durable-workflow-payload-completion") + .map(|header| super::activity_completion_uploads::Context::parse_bytes(header.as_bytes())) + .transpose()?; if declared_size(request.headers())? as u64 > runtime.external.max_bytes.max(1) { return Err(refuse( StatusCode::PAYLOAD_TOO_LARGE, @@ -233,6 +235,11 @@ async fn upload_inner(runtime: Runtime, request: Request, deadline: Instant) -> )); } let expected = metadata(request.headers(), &format!("ep_{}", ulid::Ulid::new()))?; + if let Some(context) = &context { + runtime + .authorize_completion_upload(context, &expected) + .await?; + } let timed_out = Arc::new(AtomicBool::new(false)); let timeout_flag = timed_out.clone(); let stream = request @@ -259,6 +266,13 @@ async fn upload_inner(runtime: Runtime, request: Request, deadline: Instant) -> )); } let snapshot = snapshot.map_err(snapshot_error)?; + if let Some(context) = &context + && let Some(reference) = runtime + .authorize_completion_upload(context, &expected) + .await? + { + return Ok(upload_response(reference)); + } let local = local(&runtime).await?; let quota = runtime.external.quota(runtime.namespace())?; let expires = chrono::Duration::try_seconds(runtime.external.abandoned_seconds.max(1)) @@ -269,12 +283,34 @@ async fn upload_inner(runtime: Runtime, request: Request, deadline: Instant) -> "external_payload_unavailable", ) })?; - let reference = runtime - .upload_external_local(local, expected, snapshot, expires, quota) - .await?; - Ok((StatusCode::CREATED, [("cache-control", "no-store, private")], Json(json!({ + let reference = if let Some(context) = context { + runtime + .upload_activity_completion( + local, + expected, + snapshot, + expires, + quota, + context, + runtime + .external + .completion_max_bytes + .unwrap_or(runtime.external.max_bytes), + runtime.external.abandoned_seconds, + ) + .await? + } else { + runtime + .upload_external_local(local, expected, snapshot, expires, quota) + .await? + }; + Ok(upload_response(reference)) +} + +fn upload_response(reference: Reference) -> Response { + (StatusCode::CREATED, [("cache-control", "no-store, private")], Json(json!({ "schema":"durable-workflow.v2.runtime-external-payload-upload.v1","transport_version":1,"reference":reference - }))).into_response()) + }))).into_response() } pub(super) async fn fetch( diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 7920407f0..6df259fbd 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -235,21 +235,27 @@ async fn guard(State(runtime): State, mut request: Request, next: Next) ) .await; }; - let allowed_roles: &[&str] = - if payload || path == "/api/worker/register" || path == "/api/cluster/info" { - &["worker", "operator", "admin"] - } else if worker { - &["worker"] - } else if path.starts_with("/api/workers/") - || path == "/api/namespaces" && request.method() == Method::POST - || path.starts_with("/api/namespaces/") - && path.ends_with("/external-storage") - && request.method() == Method::PUT - { - &["admin"] - } else { - &["operator", "admin"] - }; + let allowed_roles: &[&str] = if path == "/api/external-payloads/v1" + && request.method() == Method::POST + && request + .headers() + .contains_key("x-durable-workflow-payload-completion") + { + &["worker"] + } else if payload || path == "/api/worker/register" || path == "/api/cluster/info" { + &["worker", "operator", "admin"] + } else if worker { + &["worker"] + } else if path.starts_with("/api/workers/") + || path == "/api/namespaces" && request.method() == Method::POST + || path.starts_with("/api/namespaces/") + && path.ends_with("/external-storage") + && request.method() == Method::PUT + { + &["admin"] + } else { + &["operator", "admin"] + }; if !legacy_full_access && !allowed_roles.contains(&role) { return admission_error(request, worker, StatusCode::FORBIDDEN, json!({ "reason":"forbidden", "message":"Authenticated role is not allowed to access this endpoint.", diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 0e64a5caa..8539db8e5 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -2,6 +2,7 @@ //! arguments use the official codec on a bounded blocking worker. //! Typed database adapters share transitions; existing PHP data requires qualification. +mod activity_completion_uploads; mod activity_deadlines; mod activity_failures; mod activity_heartbeats; @@ -11,6 +12,7 @@ mod children; mod continuations; mod control_plane; mod cooperative; +mod cross_kind_poll; mod execution; mod external_config; mod external_registry; diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index ebcaa6c49..34040bc59 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -293,6 +293,11 @@ where .filter(|v| *v <= 60) .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?; let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); + let mut cross_kind = if timeout > 0 { + self.poll_wake_snapshot(&body, "query").await? + } else { + None + }; loop { let wake = self.wake.notified(); if let Some(task) = self.claim_query(&body).await? { @@ -305,6 +310,15 @@ where json!({"task":null,"poll_status":"timeout","server_capabilities":super::store::capabilities()}), ); } + if self + .poll_queue_changed(&body, "query", &mut cross_kind) + .await? + && tokio::time::Instant::now() < deadline + { + return Ok( + json!({"task":null,"poll_status":"task_queue_changed","server_capabilities":super::store::capabilities()}), + ); + } tokio::select! { _ = wake => {}, _ = tokio::time::sleep(Duration::from_millis(50)) => {} } } } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 75fcc4272..6e739078c 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -895,6 +895,11 @@ where "invalid_page_size", )); } + let mut cross_kind = if timeout > 0 { + self.poll_wake_snapshot(&body, kind).await? + } else { + None + }; loop { // Register the notification before probing; periodic probing also // observes work committed by an independent process. @@ -909,6 +914,15 @@ where json!({"task": null, "poll_status": "timeout", "server_capabilities": capabilities()}), ); } + if self + .poll_queue_changed(&body, kind, &mut cross_kind) + .await? + && tokio::time::Instant::now() < deadline + { + return Ok( + json!({"task":null,"poll_status":"task_queue_changed","server_capabilities":capabilities()}), + ); + } tokio::select! { _ = wake => {}, _ = tokio::time::sleep(Duration::from_millis(50)) => {} } } } diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index ac1882a09..3a325453d 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -70,6 +70,9 @@ mod condition_occurrences; #[path = "execution/query_arguments.rs"] mod query_arguments; +#[path = "execution/cross_kind_poll.rs"] +mod cross_kind_poll; + #[path = "execution/external_payloads.rs"] mod external_payloads; diff --git a/rust/tests/execution/cross_kind_poll.rs b/rust/tests/execution/cross_kind_poll.rs new file mode 100644 index 000000000..754af04ff --- /dev/null +++ b/rust/tests/execution/cross_kind_poll.rs @@ -0,0 +1,319 @@ +use super::*; +use std::time::Duration; + +async fn call(app: &Router, namespace: &str, path: &str, body: Value) -> (StatusCode, Value) { + // The default namespace deliberately uses 1.19: this capability has no + // protocol floor. Named namespaces independently exercise 1.20. + let (status, _, response) = admission::admission_http( + app, + "POST", + path, + path.starts_with("/api/worker/"), + Some("test-token"), + Some(if path.starts_with("/api/worker/") { + if namespace == "default" { + "1.19" + } else { + "1.20" + } + } else { + "2" + }), + Some(namespace), + body, + ) + .await; + (status, response) +} + +async fn register_mixed(app: &Router, namespace: &str, opt_in: bool) { + let mut capabilities = vec!["query_tasks"]; + if opt_in { + capabilities.push("cross_kind_poll_wake"); + } + let result = call( + app, + namespace, + "/api/worker/register", + json!({ + "worker_id":"mixed", "task_queue":"test", "runtime":"php", + "supported_workflow_types":["echo"], "supported_activity_types":["echo_activity"], + "capabilities":capabilities, "workflow_command_contracts":{"echo":{"queries":["state"], + "query_contracts":[{"name":"state","parameters":[]}]}} + }), + ) + .await; + assert_eq!(result.0, StatusCode::CREATED, "{}", result.1); +} + +async fn hold( + app: &Router, + namespace: &str, + kind: &str, + timeout: u64, +) -> tokio::task::JoinHandle<(StatusCode, Value)> { + let app = app.clone(); + let namespace = namespace.to_owned(); + let path = format!("/api/worker/{kind}-tasks/poll"); + let poll = tokio::spawn(async move { + call( + &app, + &namespace, + &path, + json!({"worker_id":"mixed","task_queue":"test","timeout_seconds":timeout}), + ) + .await + }); + tokio::time::sleep(Duration::from_millis(150)).await; + assert!( + !poll.is_finished(), + "the producer must act while the request remains held" + ); + poll +} + +fn hint(result: (StatusCode, Value)) { + assert_eq!(result.0, StatusCode::OK, "{}", result.1); + assert_eq!( + result.1["poll_status"], "task_queue_changed", + "{}", + result.1 + ); + assert_eq!(result.1.get("task"), Some(&Value::Null)); + assert!(result.1.get("lease_owner").is_none()); +} + +async fn admitted(app: &Router, namespace: &str, queue: &str, workflow: &str) -> Value { + let result = call( + app, + namespace, + "/api/workflows", + json!({"workflow_id":workflow, + "workflow_type":"echo","task_queue":queue,"input":envelope(Payload::Array(vec![]))}), + ) + .await; + assert_eq!(result.0, StatusCode::CREATED, "{}", result.1); + result.1 +} + +#[tokio::test] +async fn remote_workflow_admission_interrupts_both_other_kinds_without_a_lease() { + for namespace in ["default", "alpha"] { + for kind in ["activity", "query"] { + let database = TestDatabase::new().await; + let first = database.open().await.unwrap(); + let second = database.open().await.unwrap(); + let app = router(first.clone()); + let producer = router(second.clone()); + if namespace != "default" { + namespaces::create(&app, namespace).await; + } + register_mixed(&app, namespace, true).await; + let held = hold(&app, namespace, kind, 2).await; + let started = admitted(&producer, namespace, "test", "wake-original").await; + hint(held.await.unwrap()); + let leased = call( + &app, + namespace, + "/api/worker/workflow-tasks/poll", + json!({ + "worker_id":"mixed","task_queue":"test","timeout_seconds":0 + }), + ) + .await; + assert_eq!(leased.0, StatusCode::OK, "{}", leased.1); + assert_eq!(leased.1["task"]["run_id"], started["run_id"]); + assert_eq!(leased.1["task"]["workflow_task_attempt"], 1); + assert_eq!(call(&app, namespace, + &format!("/api/worker/workflow-tasks/{}/complete", leased.1["task"]["task_id"].as_str().unwrap()), + completion(&leased.1["task"],json!([{"type":"complete_workflow","result":envelope(Payload::Long(9007199254740993))}]))).await.0, + StatusCode::OK); + first.close().await; + second.close().await; + database.remove().await; + } + } +} + +#[tokio::test] +async fn scoped_hints_require_original_opt_in_and_a_new_ready_identity() { + for scenario in [ + "foreign-queue-case", + "foreign-queue-accent", + "no-opt-in", + "foreign-namespace", + "foreign-queue", + "replacement", + "preexisting", + ] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + namespaces::create(&app, "alpha").await; + register_mixed(&app, "default", scenario != "no-opt-in").await; + if scenario == "preexisting" { + admitted(&app, "default", "test", "preexisting").await; + } + let held = hold(&app, "default", "activity", 1).await; + if scenario == "replacement" { + register_mixed(&app, "default", true).await; + } + admitted( + &app, + if scenario == "foreign-namespace" { + "alpha" + } else { + "default" + }, + match scenario { + "foreign-queue-case" => "Test", + "foreign-queue-accent" => "t茅st", + "foreign-queue" | "preexisting" => "elsewhere", + _ => "test", + }, + "unrelated", + ) + .await; + let result = held.await.unwrap(); + assert_eq!(result.0, StatusCode::OK, "{scenario}: {}", result.1); + assert_eq!( + result.1["poll_status"], "timeout", + "{scenario}: {}", + result.1 + ); + assert_eq!(result.1.get("task"), Some(&Value::Null)); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn activity_admission_wakes_other_polls_but_same_kind_claim_wins() { + for kind in ["workflow", "query"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register_mixed(&app, "default", true).await; + start(&app, "activity-wake").await; + let initial = poll(&app, "mixed", "workflow").await; + let held = hold(&app, "default", kind, 2).await; + assert_eq!( + finish_task( + &app, + &initial, + json!([{"type":"schedule_activity","activity_type":"echo_activity", + "arguments":envelope(Payload::Array(vec![]))}]) + ) + .await + .0, + StatusCode::OK + ); + hint(held.await.unwrap()); + let activity = poll(&app, "mixed", "activity").await; + assert!(activity["activity_attempt_id"].is_string()); + let held = hold(&app, "default", "workflow", 2).await; + assert_eq!(request(&app,"POST", &format!("/api/worker/activity-tasks/{}/complete",activity["task_id"].as_str().unwrap()), + json!({"lease_owner":"mixed","activity_attempt_id":activity["activity_attempt_id"],"result":envelope(Payload::Null)})).await.0,StatusCode::OK); + let result = held.await.unwrap(); + assert_eq!(result.0, StatusCode::OK, "{}", result.1); + assert_eq!(result.1["poll_status"], "task"); + assert!(result.1["task"].is_object()); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn query_admission_wakes_workflow_and_activity_without_leasing_query() { + for kind in ["workflow", "activity"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register_mixed(&app, "default", true).await; + let started = start(&app, "query-wake").await; + let initial = poll(&app, "mixed", "workflow").await; + assert_eq!( + finish_task( + &app, + &initial, + json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]) + ) + .await + .0, + StatusCode::OK + ); + let prefix = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let held = hold(&app, "default", kind, 2).await; + let query = { + let app = app.clone(); + tokio::spawn(async move { + request( + &app, + "POST", + "/api/workflows/query-wake/query/state", + json!({"input":envelope(Payload::Array(vec![]))}), + ) + .await + }) + }; + let wake = held.await.unwrap(); + if query.is_finished() { + panic!( + "query ended before its worker claim: {:?}", + query.await.unwrap() + ); + } + hint(wake); + let task = poll(&app, "mixed", "query").await; + assert_eq!(task["query_task_attempt"], 1); + let result = envelope(Payload::Long(9007199254740993)); + assert_eq!( + request( + &app, + "POST", + &format!( + "/api/worker/query-tasks/{}/complete", + task["query_task_id"].as_str().unwrap() + ), + json!({"lease_owner":"mixed","query_task_attempt":1,"result_envelope":result}) + ) + .await + .0, + StatusCode::OK + ); + let reply = query.await.unwrap(); + assert_eq!(reply.0, StatusCode::OK, "{}", reply.1); + assert_eq!(reply.1["result_envelope"], result); + assert_eq!( + run_history(&app, &started["workflow_id"], &started["run_id"]).await, + prefix + ); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn zero_timeout_polls_keep_the_existing_empty_reply() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register_mixed(&app, "default", true).await; + admitted(&app, "default", "test", "already-ready").await; + for kind in ["activity", "query"] { + let result = call( + &app, + "default", + &format!("/api/worker/{kind}-tasks/poll"), + json!({ + "worker_id":"mixed","task_queue":"test","timeout_seconds":0 + }), + ) + .await; + assert_eq!(result.0, StatusCode::OK, "{}", result.1); + assert_eq!(result.1["poll_status"], "timeout"); + assert_eq!(result.1.get("task"), Some(&Value::Null)); + } + runtime.close().await; + database.remove().await; +} diff --git a/rust/tests/execution/external_payloads.rs b/rust/tests/execution/external_payloads.rs index 31360d5fe..22b54ed9c 100644 --- a/rust/tests/execution/external_payloads.rs +++ b/rust/tests/execution/external_payloads.rs @@ -1,4 +1,6 @@ use super::*; +#[path = "external_payloads/completion_uploads.rs"] +mod completion_uploads; use axum::{body::Bytes, http::HeaderMap}; use durable_workflow_server::runtime::ExternalPayloadConfig; use sha2::{Digest, Sha256}; @@ -1310,8 +1312,31 @@ async fn authentication_namespace_and_declared_limit_refuse_before_body_or_provi assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); let body: Value = serde_json::from_slice(&to_bytes(response.into_body(), 4096).await.unwrap()).unwrap(); + assert_eq!(body["reason"], "external_payload_completion_invalid"); + let unsupported = json!({"schema":"durable-workflow.v2.payload-completion-context.v1", + "kind":"workflow","task_id":"task","attempt":1,"lease_owner":"owner", + "operation":"complete","slot":["commands",0,"result"]}); + let response = router + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/external-payloads/v1") + .header("authorization", "Bearer worker-token") + .header("content-type", "application/octet-stream") + .header( + "x-durable-workflow-payload-completion", + unsupported.to_string(), + ) + .body(untouched(reads.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); assert_eq!( - body["reason"], + serde_json::from_slice::(&bytes).unwrap()["reason"], "development_payload_completion_upload_unqualified" ); assert_eq!(reads.load(Ordering::Relaxed), 0); @@ -1323,14 +1348,17 @@ async fn authentication_namespace_and_declared_limit_refuse_before_body_or_provi #[tokio::test] async fn stalled_upload_keeps_original_timeout_and_releases_capacity_for_next_original() { let temp = tempfile::tempdir().unwrap(); - let mut config = ExternalPayloadConfig::default(); + let recovery_config = ExternalPayloadConfig::default(); + // The clone shares the one-slot worker semaphore. Keep the stalled + // request's deadline separate from fresh durable filesystem/SQL work. + let mut config = recovery_config.clone(); config.timeout = Duration::from_millis(100); - let (database, runtime, router) = setup(config, temp.path()).await; + let (database, runtime, stalled_router) = setup(config, temp.path()).await; let meta = metadata(b"abc"); let start = tokio::time::Instant::now(); let body = Body::from_stream(futures_util::stream::pending::>()); let (status, _, _) = request( - &router, + &stalled_router, "POST", "/api/external-payloads/v1", "worker-token", @@ -1341,8 +1369,15 @@ async fn stalled_upload_keeps_original_timeout_and_releases_capacity_for_next_or .await; assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); assert!(start.elapsed() < Duration::from_secs(1)); + // This fresh request uses the normal policy with the same capacity pool. + // A leaked permit still times out; slow durable I/O is not a leak proof. + let recovery_router = router( + runtime + .clone() + .with_external_payload_config(recovery_config), + ); let (status, _, body) = request( - &router, + &recovery_router, "POST", "/api/external-payloads/v1", "worker-token", diff --git a/rust/tests/execution/external_payloads/completion_uploads.rs b/rust/tests/execution/external_payloads/completion_uploads.rs new file mode 100644 index 000000000..2c4a80878 --- /dev/null +++ b/rust/tests/execution/external_payloads/completion_uploads.rs @@ -0,0 +1,448 @@ +use super::*; + +fn provider_files(path: &std::path::Path) -> usize { + if !path.exists() { + return 0; + } + std::fs::read_dir(path) + .unwrap() + .map(|entry| { + let path = entry.unwrap().path(); + if path.is_dir() { + provider_files(&path) + } else { + 1 + } + }) + .sum() +} + +async fn upload( + app: &Router, + token: &str, + namespace: &str, + context: &Value, + bytes: Vec, + body: Option, +) -> (StatusCode, HeaderMap, Value) { + let request = Request::builder() + .method("POST") + .uri("/api/external-payloads/v1") + .header("authorization", format!("Bearer {token}")) + .header("x-namespace", namespace) + .header("content-type", "application/octet-stream") + .header("x-durable-workflow-payload-codec", "avro") + .header("x-durable-workflow-payload-size", bytes.len().to_string()) + .header("x-durable-workflow-payload-sha256", digest(&bytes)) + .header("x-durable-workflow-payload-completion", context.to_string()) + .body(body.unwrap_or_else(|| Body::from(bytes))) + .unwrap(); + let response = app.clone().oneshot(request).await.unwrap(); + let status = response.status(); + let headers = response.headers().clone(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + (status, headers, serde_json::from_slice(&bytes).unwrap()) +} + +async fn original( + start_to_close: i64, +) -> ( + tempfile::TempDir, + TestDatabase, + Runtime, + Router, + Value, + Vec, +) { + let temp = tempfile::tempdir().unwrap(); + let mut config = ExternalPayloadConfig::default(); + config.completion_max_bytes = Some(64); + config.timeout = Duration::from_secs(5); + let (database, runtime, app) = setup(config, &temp.path().join("objects")).await; + let registered = json_request( + &app, + "POST", + "/api/worker/register", + "worker-token", + json!({ + "worker_id":"completion-worker/位馃榾", "task_queue":"test", "runtime":"php", + "supported_workflow_types":["echo"], "supported_activity_types":["echo_activity"] + }), + ) + .await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let admitted = json_request(&app, "POST", "/api/workflows", "operator-token", json!({ + "workflow_id":"completion-upload", "workflow_type":"echo", "task_queue":"test", + "input":envelope(Payload::Array(vec![])), "run_timeout_seconds":120, "execution_timeout_seconds":600 + })).await; + assert_eq!(admitted.0, StatusCode::CREATED, "{}", admitted.1); + let poll = + json!({"worker_id":"completion-worker/位馃榾", "task_queue":"test", "timeout_seconds":0}); + let task = json_request( + &app, + "POST", + "/api/worker/workflow-tasks/poll", + "worker-token", + poll.clone(), + ) + .await + .1["task"] + .clone(); + let scheduled = json_request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), "worker-token", completion(&task, json!([{ + "type":"schedule_activity", "activity_type":"echo_activity", "arguments":envelope(Payload::Array(vec![])), + "retry_policy":{"max_attempts":1,"backoff_seconds":[]}, "start_to_close_timeout":start_to_close, + "schedule_to_close_timeout":600 + }]))).await; + assert_eq!(scheduled.0, StatusCode::OK, "{}", scheduled.1); + let task = json_request( + &app, + "POST", + "/api/worker/activity-tasks/poll", + "worker-token", + poll, + ) + .await + .1["task"] + .clone(); + assert_eq!(task["run_id"], admitted.1["run_id"]); + let bytes = ValueCodec::new() + .unwrap() + .encode(&Payload::Array(vec![ + Payload::Long(9007199254740993), + Payload::Null, + Payload::String("range位".into()), + ])) + .unwrap() + .into_bytes(); + (temp, database, runtime, app, task, bytes) +} + +fn context(task: &Value) -> Value { + json!({"schema":"durable-workflow.v2.payload-completion-context.v1", "kind":"activity", + "task_id":task["task_id"], "attempt":task["activity_attempt_id"], "lease_owner":task["lease_owner"], + "operation":"complete", "slot":["result"]}) +} + +#[tokio::test] +async fn activity_completion_uploads_preserve_authority_allowance_reference_and_cold_parent() { + let (_provider, database, runtime, app, task, bytes) = original(120).await; + let context = context(&task); + let mut utf8 = context.clone(); + utf8["lease_owner"] = json!("foreign/位馃榾"); + let refused = upload(&app, "worker-token", "default", &utf8, bytes.clone(), None).await; + assert_eq!( + refused.0, + StatusCode::CONFLICT, + "valid UTF-8 context: {}", + refused.2 + ); + assert_eq!( + refused.2["reason"], + "external_payload_completion_lease_rejected" + ); + let named = json_request( + &app, + "POST", + "/api/namespaces", + "admin-token", + json!({"name":"other"}), + ) + .await; + assert_eq!(named.0, StatusCode::CREATED, "{}", named.1); + let foreign = upload(&app, "worker-token", "other", &context, bytes.clone(), None).await; + assert_eq!(foreign.0, StatusCode::CONFLICT, "{}", foreign.2); + assert_eq!( + foreign.2["reason"], + "external_payload_completion_lease_rejected" + ); + for role in ["operator-token", "admin-token"] { + assert_eq!( + upload(&app, role, "default", &context, bytes.clone(), None) + .await + .0, + StatusCode::FORBIDDEN + ); + } + for field in ["task_id", "attempt", "lease_owner"] { + let read = Arc::new(AtomicUsize::new(0)); + let observed = read.clone(); + let body = Body::from_stream(futures_util::stream::once(async move { + observed.fetch_add(1, Ordering::SeqCst); + Ok::<_, std::io::Error>(Bytes::from_static(b"unread")) + })); + let mut bad = context.clone(); + bad[field] = json!("foreign"); + let response = upload( + &app, + "worker-token", + "default", + &bad, + bytes.clone(), + Some(body), + ) + .await; + assert_eq!(response.0, StatusCode::CONFLICT, "{}", response.2); + assert_eq!( + response.2["reason"], + "external_payload_completion_lease_rejected" + ); + assert_eq!( + read.load(Ordering::SeqCst), + 0, + "fenced context must not demand provider bytes" + ); + } + assert_eq!(unclaimed(&database).await, 0); + for field in ["task_id", "attempt"] { + let mut alias = context.clone(); + let original = alias[field].as_str().unwrap(); + let lower = original.to_ascii_lowercase(); + let changed = if lower == original { + original.to_ascii_uppercase() + } else { + lower + }; + assert_ne!(changed, original); + alias[field] = json!(changed); + let response = upload(&app, "worker-token", "default", &alias, bytes.clone(), None).await; + assert_eq!( + response.0, + StatusCode::CONFLICT, + "noncanonical {field}: {}", + response.2 + ); + assert_eq!( + response.2["reason"], + "external_payload_completion_lease_rejected" + ); + } + let budget = upload( + &app, + "worker-token", + "default", + &context, + vec![b'x'; 65], + None, + ) + .await; + assert_eq!(budget.0, StatusCode::TOO_MANY_REQUESTS, "{}", budget.2); + assert_eq!( + budget.2["reason"], + "external_payload_completion_budget_exhausted" + ); + assert_eq!(budget.2["retry_after_seconds"], 5); + assert_eq!(budget.1["retry-after"], "5"); + assert_eq!(unclaimed(&database).await, 0); + let uploaded = upload( + &app, + "worker-token", + "default", + &context, + bytes.clone(), + None, + ) + .await; + assert_eq!(uploaded.0, StatusCode::CREATED, "{}", uploaded.2); + let reference = uploaded.2["reference"].clone(); + assert_eq!(reference["sha256"], digest(&bytes)); + let duplicate = upload( + &app, + "worker-token", + "default", + &context, + bytes.clone(), + None, + ) + .await; + assert_eq!(duplicate.0, StatusCode::CREATED); + assert_eq!(duplicate.2["reference"], reference); + let conflict = upload( + &app, + "worker-token", + "default", + &context, + b"changed".to_vec(), + None, + ) + .await; + assert_eq!(conflict.0, StatusCode::CONFLICT); + assert_eq!(conflict.2["reason"], "external_payload_completion_conflict"); + assert_eq!(unclaimed(&database).await, 1); + let completed = json_request( + &app, + "POST", + &format!( + "/api/worker/activity-tasks/{}/complete", + task["task_id"].as_str().unwrap() + ), + "worker-token", + json!({ + "activity_attempt_id":task["activity_attempt_id"], "lease_owner":task["lease_owner"], + "result":{"codec":"avro","external_payload":reference} + }), + ) + .await; + assert_eq!(completed.0, StatusCode::OK, "{}", completed.1); + assert_eq!(completed.1["recorded"], true); + assert_eq!(unclaimed(&database).await, 0); + let duplicate = upload( + &app, + "worker-token", + "default", + &context, + bytes.clone(), + None, + ) + .await; + assert_eq!(duplicate.0, StatusCode::CREATED, "{}", duplicate.2); + assert_eq!(duplicate.2["reference"], reference); + let cold = database.open().await.unwrap().with_role_tokens( + Some("worker-token".into()), + Some("operator-token".into()), + Some("admin-token".into()), + ); + let cold_app = router(cold.clone()); + let parent = json_request( + &cold_app, + "POST", + "/api/worker/workflow-tasks/poll", + "worker-token", + json!({ + "worker_id":"completion-worker/位馃榾", "task_queue":"test", "timeout_seconds":0 + }), + ) + .await + .1["task"] + .clone(); + assert_eq!(parent["run_id"], task["run_id"]); + let events = parent["history_events"] + .as_array() + .unwrap() + .iter() + .filter(|event| event["event_type"] == "ActivityCompleted") + .collect::>(); + assert_eq!(events.len(), 1); + assert_eq!( + events[0]["payload"]["result"]["external_payload"], + reference + ); + let fetched = request( + &cold_app, + "GET", + &format!( + "/api/external-payloads/v1/{}", + reference["reference_id"].as_str().unwrap() + ), + "worker-token", + "default", + Some(&reference), + Body::empty(), + ) + .await; + assert_eq!(fetched.0, StatusCode::OK); + assert_eq!(fetched.2, bytes); + let result = ValueCodec::new() + .unwrap() + .decode(std::str::from_utf8(&fetched.2).unwrap()) + .unwrap(); + assert_eq!( + result, + Payload::Array(vec![ + Payload::Long(9007199254740993), + Payload::Null, + Payload::String("range位".into()) + ]) + ); + let final_result = json_request( + &cold_app, + "POST", + &format!( + "/api/worker/workflow-tasks/{}/complete", + parent["task_id"].as_str().unwrap() + ), + "worker-token", + completion( + &parent, + json!([{"type":"complete_workflow","result":envelope(result)}]), + ), + ) + .await; + assert_eq!(final_result.0, StatusCode::OK, "{}", final_result.1); + cold.close().await; + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn activity_expiry_during_upload_leaves_no_ready_reference_or_provider_write() { + let (provider, database, runtime, app, task, bytes) = original(1).await; + let body_bytes = bytes.clone(); + let body = Body::from_stream(futures_util::stream::once(async move { + tokio::time::sleep(Duration::from_millis(2500)).await; + Ok::<_, std::io::Error>(Bytes::from(body_bytes)) + })); + let refused = upload( + &app, + "worker-token", + "default", + &context(&task), + bytes, + Some(body), + ) + .await; + assert_eq!(refused.0, StatusCode::CONFLICT, "{}", refused.2); + assert_eq!( + refused.2["reason"], + "external_payload_completion_lease_rejected" + ); + assert_eq!(unclaimed(&database).await, 0); + assert_eq!(provider_files(provider.path()), 0); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn activity_completion_during_upload_cannot_reopen_its_closed_allowance() { + let (provider, database, runtime, app, task, bytes) = original(120).await; + let entered = Arc::new(tokio::sync::Notify::new()); + let observed = entered.clone(); + let (release, wait) = tokio::sync::oneshot::channel(); + let body_bytes = bytes.clone(); + let body = Body::from_stream(futures_util::stream::once(async move { + observed.notify_one(); + wait.await.unwrap(); + Ok::<_, std::io::Error>(Bytes::from(body_bytes)) + })); + let upload_app = app.clone(); + let original_context = context(&task); + let pending = tokio::spawn(async move { + upload( + &upload_app, + "worker-token", + "default", + &original_context, + bytes, + Some(body), + ) + .await + }); + tokio::time::timeout(Duration::from_secs(3), entered.notified()) + .await + .unwrap(); + let complete = json_request(&app, "POST", &format!("/api/worker/activity-tasks/{}/complete", task["task_id"].as_str().unwrap()), "worker-token", json!({ + "activity_attempt_id":task["activity_attempt_id"],"lease_owner":task["lease_owner"],"result":envelope(Payload::Null) + })).await; + assert_eq!(complete.0, StatusCode::OK, "{}", complete.1); + assert_eq!(complete.1["recorded"], true); + release.send(()).unwrap(); + let refused = pending.await.unwrap(); + assert_eq!(refused.0, StatusCode::CONFLICT, "{}", refused.2); + assert_eq!( + refused.2["reason"], + "external_payload_completion_lease_rejected" + ); + assert_eq!(unclaimed(&database).await, 0); + assert_eq!(provider_files(provider.path()), 0); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/ACTIVITY-COMPLETION-UPLOADS.md b/scripts/conformance/server-parity/ACTIVITY-COMPLETION-UPLOADS.md new file mode 100644 index 000000000..f64ba17e3 --- /dev/null +++ b/scripts/conformance/server-parity/ACTIVITY-COMPLETION-UPLOADS.md @@ -0,0 +1,29 @@ +# Activity completion uploads + +This #325 slice implements the existing published version-1 activity-result +completion context on the native local byte transport. The unchanged native +control refuses the header. The existing wire contract is in +`resources/platform-protocol-specs/external-payload-transport.openapi.yaml` and +`docs/storage-admission.md`. PHP remains the published reference. + +Acceptance requires actual separate PHP/native originals, a namespace-bound +current activity attempt before body/provider work, an allowance shared by that +lease, conflicting slot refusal and rechecks after transfer. Upload does not +complete or renew the task. Matching recorded upload references may reconcile +after closure without rewriting or extending provider state. Final activity +completion still applies its own atomic authority fence. + +Qualification covers activity result uploads, actual result +completion and cold parent consumption, changed-owner/attempt/namespace and +expired-lease refusal, duplicate/conflicting bytes, bounded allowances and +expiry during transfer. Raw UTF-8 and escaped JSON contexts preserve the same +identity; case-folding database aliases cannot create another allowance. +Published PHP/SDK tuples and earlier failed originals +stay unchanged. All database families require qualification before promotion. + +Workflow/query contexts, activity failure details, prepared version-2 contexts, +storage-pressure bypass, S3, restart/cleanup/backup faults and full transport +discovery remain separate gates. Exact-source local controls and genuine +published PHP/native consumer outcomes are recorded in #458; all-six hosted +qualification and promotion remain pending. No advertised capability or release +follows from the local checks. diff --git a/scripts/conformance/server-parity/CROSS-KIND-POLL-WAKE.md b/scripts/conformance/server-parity/CROSS-KIND-POLL-WAKE.md new file mode 100644 index 000000000..998225dfa --- /dev/null +++ b/scripts/conformance/server-parity/CROSS-KIND-POLL-WAKE.md @@ -0,0 +1,94 @@ +# Optional worker poll wake + +Published PHP already supports the worker registration capability +`cross_kind_poll_wake`. The current OpenAPI documents its existing +`task=null, poll_status=task_queue_changed` response. This is a capability-only +opt-in. It adds no worker-protocol floor and changes no frozen schema or actor. +The agreed boundary and actual published/native observations are recorded in +[Server #325](https://github.com/durable-workflow/server/issues/325#issuecomment-6178479885) +and the [PHP contract acknowledgment](https://github.com/durable-workflow/server/issues/325#issuecomment-6178506403). + +The response asks the worker to sweep its task kinds again. It supplies no +task, creates no lease, accepts no command and grants no execution authority. +Normal authentication, registration, namespace, queue, routing and claim fences +still apply to each subsequent request. A wake does not guarantee that a task +remains available when that worker polls again. It does not renew a poll, +workflow, activity, query or cleanup deadline. + +The hint is scoped to the registered namespace and task queue. An unrelated +namespace, queue or global notification grants no authority to emit it. +Workers that do not opt in retain their existing behavior, including other +already-defined empty or blocked statuses. + +The first real worker controls use the explicit default namespace. They park an +ordinary installed mixed worker in an activity poll or a query poll, then admit +one normal one-activity workflow while +that request remains unanswered. The PHP reference interrupts the held request +with the explicit hint. Its worker follows with a workflow sweep and completes +the original. The unchanged native source instead consumes the remaining +other-kind waits. Both still complete exact outputs and ordered histories. +These observations identify the controlled parity gap. They are not a matched +latency comparison and do not explain every prior performance tail. + +The reusable observer uses the unchanged capacity definitions and normal +installed SDK selected by `benchmarks/server-port/v2/matched.json`. Set +`DW_PARITY_SDK_AUTOLOAD` to that installation's autoloader and +`DW_PARITY_TOKEN` to the isolated target's test token. Give each backend a +separate fresh database and owned output directory. Then run in the repository +runtime: + +```sh +profile=benchmarks/server-port/v2/matched.json +php scripts/conformance/server-parity/cross-kind-poll-observe.php \ + "$runtime_url" "$unique_prefix" "$profile" "$output_directory" \ + > "$output_directory/observations.ndjson" +node scripts/conformance/server-parity/cross-kind-poll-check.mjs \ + "$profile" "$output_directory/observations.ndjson" +``` + +Observer exit 0 proves completed observations, not wake acceptance. The separate +reader requires the actual null-task hint, same worker/queue/namespace, a +zero-timeout first-attempt claim of the original workflow/run, exact output/history and the original 35-second +budget. Constructed reader controls refuse ordinary timeout completion and +other malformed observations. Current qualification is recorded in +[PR #455](https://github.com/durable-workflow/server/pull/455). +These controls are not selected in default fixtures. + +Native qualification must cover those two cases, opt-in refusal boundaries, +namespace/queue isolation, ordinary same-kind claims and all database families. +Matched installed-worker controls for other task-kind pairs and registration +replacement remain explicit follow-up gates. Preserve raw statuses, request/response ordering, durable identities and +original deadlines. Completion alone does not prove the wake behavior. A reader +must verify the actual null-task hint and the subsequent workflow poll. + +The native correction observes newly ready task/query IDs in the exact +namespace and queue. It returns a hint only after the ordinary same-kind claim +finds no task. Database probes cover independent nodes; the global in-process +notification merely schedules another probe. Initial availability is the +baseline, so preexisting work cannot create a busy loop of hints. Zero-timeout +polls keep their original behavior. A changed registration incarnation or +removed opt-in disables hints for that held request. No lease, task status, +receipt, history or deadline is mutated by these observations. + +MySQL task candidates keep the indexed scope predicate and then compare exact +routing bytes. The preserved collation otherwise treats `test`, `Test` and +`t茅st` as equal. Query metadata uses binary namespace/queue comparisons. + +Task observations keep at most 100 IDs; query observations keep at most 65. +The query projection reads metadata inside the database and does not transfer +history snapshots to the HTTP executor. These are conservative availability +hints. An admission already consumed by another worker need not interrupt the +held poll. The hint never promises eligibility or a remaining claimable task. + +Five focused native HTTP test groups cover all six task-kind directions, +same-kind claim precedence, default 1.19 and named-namespace 1.20 polls, +independent pools, exact queue/namespace isolation, absent opt-in, replacement, +preexisting work and zero timeout. They pass locally on SQLite, PostgreSQL 17 +and MySQL 8.4. Against unchanged +native tree `696130d7`, the three positive groups fail with ordinary timeouts; +the two refusal/zero-timeout groups pass. The complete six-cell matrix and +installed-worker coverage of other task-kind directions remain acceptance gates. + +This is a service-mode worker HTTP contract. Embedded execution has no such +worker HTTP wait and is inapplicable to this specific observation. It retains +its independent workflow/activity/replay qualification. diff --git a/scripts/conformance/server-parity/EXTERNAL-CONSUMERS.md b/scripts/conformance/server-parity/EXTERNAL-CONSUMERS.md index 00abdace4..6b6e7bd77 100644 --- a/scripts/conformance/server-parity/EXTERNAL-CONSUMERS.md +++ b/scripts/conformance/server-parity/EXTERNAL-CONSUMERS.md @@ -3,7 +3,12 @@ The profile reuses the normal PHP SDK and immutable PHP Server selected by `benchmarks/server-port/v2/matched.json`. It runs five independent originals on each of separate PHP/native databases: external workflow input, workflow result, -activity result and one/two external activity argument batches. Separate client +activity result and one/two external activity argument batches. A sixth original +uses the published raw HTTP version-1 activity-result completion-upload context, +then completes the activity and its cold parent. It checks duplicate/conflicting +uploads, real UTF-8 context JSON, foreign ownership, unchanged lease expiry and matching upload +reconciliation after activity closure. This sixth case uses the installed Core +Avro serializer rather than claiming an SDK worker callback. Separate client processes perform authorship, actual activity callbacks and cold result reads. The existing recipes keep their original 30-second bounds and exact typed int64/null/Unicode values. Explicit reference submissions remain explicit; @@ -23,12 +28,13 @@ result summaries and diagnostic logs enter the existing bounded artifact exporte The independent 64-bit PHP reader checks original identities, typed values, encoded digests, activity counts, fixed bounds and unchanged cold-read history. -Seven corruptions of the actual summaries must refuse. These reader controls +References compare every member and type exactly; JSON object member order is +observed separately and cannot change identity. Twelve corruptions of the actual summaries must refuse. These reader controls do not execute another business original or establish a database-fault claim. Embedded execution has no service HTTP payload-reference carrier in this profile; its workflow semantics remain covered by the independent embedded suite. Server restart, S3, automatic result policy, missing ingress fields, -completion-upload authority and cleanup/backup fault qualification remain +broader workflow/query/prepared completion uploads and cleanup/backup fault qualification remain separate gates. Full native external transport discovery stays unadvertised. Qualification of this profile is pending in its owning #325 pull request. diff --git a/scripts/conformance/server-parity/activity-completion-upload-probe.php b/scripts/conformance/server-parity/activity-completion-upload-probe.php new file mode 100644 index 000000000..fbd0dddad --- /dev/null +++ b/scripts/conformance/server-parity/activity-completion-upload-probe.php @@ -0,0 +1,166 @@ + rtrim($url, '/'), 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 5]); +$headers = ['Authorization' => 'Bearer '.(getenv('DW_PARITY_TOKEN') ?: 'completion-fixture-token'), 'X-Namespace' => 'default', + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => '1.20']; +$request = static function (string $method, string $path, array $body = []) use ($http, $headers, $deadline): array { + if (microtime(true) >= $deadline) { + throw new RuntimeException('Original 30-second observation expired'); + } + $response = $http->request($method, $path, ['headers' => $headers, 'json' => $body, + 'timeout' => min(5, max(0.001, $deadline - microtime(true)))]); + + return [$response->getStatusCode(), json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; +}; +$require = static function (bool $condition, string $message): void { + if (! $condition) { + throw new RuntimeException($message); + } +}; +$expect = static function (array $response, int $status, string $operation) use ($require): array { + $require($response[0] === $status, $operation.': '.json_encode([ + 'status' => $response[0], 'reason' => $response[1]['reason'] ?? null, + 'errors' => $response[1]['errors'] ?? null, + ])); + + return $response[1]; +}; +$envelope = static fn (mixed $value): array => ['codec' => 'avro', 'blob' => Avro::serialize($value)]; +$sameReference = static function (array $left, array $right): bool { + if (count($left) !== 5 || count($right) !== 5) { + return false; + } + foreach (['schema', 'reference_id', 'codec', 'size_bytes', 'sha256'] as $field) { + if (! array_key_exists($field, $left) || ! array_key_exists($field, $right) || $left[$field] !== $right[$field]) { + return false; + } + } + + return true; +}; +if ($request('GET', '/api/namespaces/default')[0] === 404) { + $expect($request('POST', '/api/namespaces', ['name' => 'default']), 201, 'namespace'); +} +$expect($request('PUT', '/api/namespaces/default/external-storage', ['driver' => 'local', 'enabled' => true, + 'threshold_bytes' => 16, 'config' => ['uri' => 'file:///state/objects']]), 200, 'policy'); +$expect($request('POST', '/api/worker/register', ['worker_id' => $workerId, 'task_queue' => $prefix, + 'runtime' => 'php', 'sdk_version' => 'raw-http-control', + 'supported_workflow_types' => ['parity.completion_upload'], 'supported_activity_types' => ['parity.completion_activity'], + 'capabilities' => [], 'capability_manifest' => array_fill_keys( + ['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http_control'], + )]), 201, 'registration'); +$original = $expect($request('POST', '/api/workflows', ['workflow_id' => $prefix, + 'workflow_type' => 'parity.completion_upload', 'task_queue' => $prefix, 'input' => $envelope([]), + 'run_timeout_seconds' => 120, 'execution_timeout_seconds' => 600]), 201, 'original admission'); +fwrite(STDERR, json_encode(['phase' => 'admitted', 'run_id' => $original['run_id'], + 'started_at' => $started, 'original_deadline' => $deadline], JSON_THROW_ON_ERROR)."\n"); +$poll = ['worker_id' => $workerId, 'task_queue' => $prefix, 'timeout_seconds' => 0]; +$task = $expect($request('POST', '/api/worker/workflow-tasks/poll', $poll), 200, 'workflow claim')['task']; +$require($task['run_id'] === $original['run_id'], 'Original workflow claim changed'); +$expect($request('POST', '/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], + 'commands' => [['type' => 'schedule_activity', 'activity_type' => 'parity.completion_activity', + 'arguments' => $envelope([]), 'retry_policy' => ['max_attempts' => 1, 'backoff_seconds' => []], + 'start_to_close_timeout' => 120, 'schedule_to_close_timeout' => 600]], +]), 200, 'activity admission'); +$activity = $expect($request('POST', '/api/worker/activity-tasks/poll', $poll), 200, 'activity claim')['task']; +$require($activity['run_id'] === $original['run_id'], 'Original activity claim changed'); +$statusPath = '/api/worker/activity-tasks/'.$activity['task_id'].'/status'; +$ownership = ['activity_attempt_id' => $activity['activity_attempt_id'], 'lease_owner' => $activity['lease_owner']]; +$beforeUpload = $expect($request('POST', $statusPath, $ownership), 200, 'original activity status'); +$require($beforeUpload['can_continue'] === true, 'Original activity lacks upload authority'); +$context = ['schema' => 'durable-workflow.v2.payload-completion-context.v1', 'kind' => 'activity', + 'task_id' => $activity['task_id'], 'attempt' => $activity['activity_attempt_id'], + 'lease_owner' => $activity['lease_owner'], 'operation' => 'complete', 'slot' => ['result']]; +$require(str_contains($context['lease_owner'], '位馃榾'), 'Completion owner lost actual UTF-8'); +$expected = [9007199254740993, null, 'range位']; +$bytes = Avro::serialize($expected); +$upload = static function (array $context, string $bytes) use ($http, $headers, $deadline): array { + if (microtime(true) >= $deadline) { + throw new RuntimeException('Original upload observation expired'); + } + $response = $http->post('/api/external-payloads/v1', ['headers' => $headers + [ + 'Content-Type' => 'application/octet-stream', 'X-Durable-Workflow-Payload-Codec' => 'avro', + 'X-Durable-Workflow-Payload-Size' => (string) strlen($bytes), + 'X-Durable-Workflow-Payload-SHA256' => hash('sha256', $bytes), + 'X-Durable-Workflow-Payload-Completion' => json_encode($context, JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE), + ], 'body' => $bytes, 'timeout' => min(5, max(0.001, $deadline - microtime(true)))]); + + return [$response->getStatusCode(), json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; +}; +$first = $upload($context, $bytes); +if ($first[0] === 422 && ($first[1]['reason'] ?? null) === 'development_payload_completion_upload_unqualified') { + echo json_encode(['outcome' => 'unsupported', 'scope' => 'native_completion_upload_counterfactual', + 'run_id' => $original['run_id'], 'task_id' => $activity['task_id'], 'response' => $first, + 'original_budget_seconds' => 30], JSON_THROW_ON_ERROR)."\n"; + exit(2); +} +$reference = $expect($first, 201, 'completion upload')['reference']; +$require($reference['sha256'] === hash('sha256', $bytes) && $reference['size_bytes'] === strlen($bytes), 'Upload bytes changed'); +$duplicate = $expect($upload($context, $bytes), 201, 'identical upload')['reference']; +$require($sameReference($duplicate, $reference), 'Duplicate upload identity changed: '.json_encode([$reference, $duplicate], JSON_THROW_ON_ERROR)); +$conflict = $upload($context, Avro::serialize('different')); +$expect($conflict, 409, 'conflicting upload'); +$require($conflict[1]['reason'] === 'external_payload_completion_conflict', 'Unexpected conflict disposition'); +$wrongOwner = $context; +$wrongOwner['lease_owner'] .= '-wrong'; +$refused = $upload($wrongOwner, $bytes); +$expect($refused, 409, 'foreign owner'); +$require($refused[1]['reason'] === 'external_payload_completion_lease_rejected', 'Unexpected owner disposition'); +$afterUpload = $expect($request('POST', $statusPath, $ownership), 200, 'activity status after uploads'); +$require($afterUpload['can_continue'] === true && $afterUpload['lease_expires_at'] === $beforeUpload['lease_expires_at'], + 'Upload changed original activity lease authority'); +$completion = $expect($request('POST', '/api/worker/activity-tasks/'.$activity['task_id'].'/complete', [ + 'lease_owner' => $activity['lease_owner'], 'activity_attempt_id' => $activity['activity_attempt_id'], + 'result' => ['codec' => 'avro', 'external_payload' => $reference], +]), 200, 'activity completion'); +$require($completion['recorded'] === true, 'Activity outcome was not committed'); +$require($sameReference($expect($upload($context, $bytes), 201, 'closed upload reconciliation')['reference'], $reference), + 'Closed upload retry changed its reference'); +$parent = $expect($request('POST', '/api/worker/workflow-tasks/poll', $poll), 200, 'parent claim')['task']; +$require($parent['run_id'] === $original['run_id'], 'Cold parent changed original'); +$completed = array_values(array_filter($parent['history_events'], static fn (array $event): bool => $event['event_type'] === 'ActivityCompleted')); +$require(count($completed) === 1, 'Activity completion history is not unique'); +$result = $completed[0]['payload']['result']; +if (isset($result['blob'])) { + $require(Avro::unserialize($result['blob']) === $expected, 'Cold inline result changed'); +} else { + $require($sameReference($result['external_payload'] ?? [], $reference), 'Cold external result changed'); + $response = $http->get('/api/external-payloads/v1/'.$reference['reference_id'], ['headers' => $headers + [ + 'X-Durable-Workflow-Payload-Codec' => 'avro', 'X-Durable-Workflow-Payload-Size' => (string) strlen($bytes), + 'X-Durable-Workflow-Payload-SHA256' => hash('sha256', $bytes), + ]]); + $require($response->getStatusCode() === 200 && (string) $response->getBody() === $bytes, 'Cold bytes changed'); +} +$expect($request('POST', '/api/worker/workflow-tasks/'.$parent['task_id'].'/complete', [ + 'lease_owner' => $parent['lease_owner'], 'workflow_task_attempt' => $parent['workflow_task_attempt'], + 'commands' => [['type' => 'complete_workflow', 'result' => $envelope($expected)]], +]), 200, 'parent completion'); +$describe = $expect($request('GET', '/api/workflows/'.$prefix.'/runs/'.$original['run_id']), 200, 'original result'); +$require($describe['status'] === 'completed' && microtime(true) < $deadline, 'Original did not complete within its budget'); +$expect($request('DELETE', '/api/worker/registrations/'.rawurlencode($workerId)), 200, 'registration retirement'); +$finished = microtime(true); +echo json_encode(['outcome' => 'pass', 'scope' => 'raw_http_activity_completion_upload', + 'run_id' => $original['run_id'], 'task_id' => $activity['task_id'], 'activity_attempt_id' => $activity['activity_attempt_id'], + 'lease_owner' => $activity['lease_owner'], + 'parent_run_id' => $parent['run_id'], 'parent_task_id' => $parent['task_id'], 'result' => $expected, + 'activity_completion_count' => count($completed), 'workflow_status' => $describe['status'], + 'lease_expires_before' => $beforeUpload['lease_expires_at'], 'lease_expires_after' => $afterUpload['lease_expires_at'], + 'reference' => $reference, 'duplicate_reference' => $duplicate, + 'duplicate_key_order_changed' => array_keys($reference) !== array_keys($duplicate), + 'original_budget_seconds' => 30, 'elapsed_seconds' => $finished - $started, + 'started_at' => $started, 'original_deadline' => $deadline, 'finished_at' => $finished, + 'controls' => ['identical_upload', 'conflicting_slot', 'foreign_owner', 'closed_upload_reconciliation', 'cold_parent', 'utf8_completion_context']], JSON_THROW_ON_ERROR)."\n"; diff --git a/scripts/conformance/server-parity/cross-kind-poll-check.mjs b/scripts/conformance/server-parity/cross-kind-poll-check.mjs new file mode 100644 index 000000000..71c29214a --- /dev/null +++ b/scripts/conformance/server-parity/cross-kind-poll-check.mjs @@ -0,0 +1,13 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkCrossKindPollWake} from './cross-kind-poll-contract.mjs'; + +const [profilePath, observationPath] = process.argv.slice(2); +if (!profilePath || !observationPath || process.argv.length !== 4) { + throw new Error('Usage: cross-kind-poll-check.mjs PROFILE_JSON OBSERVATIONS_NDJSON'); +} +const profile = JSON.parse(readFileSync(profilePath, 'utf8')); +const records = readFileSync(observationPath, 'utf8').trim().split('\n').map(line => JSON.parse(line)); +assert.deepEqual(records.map(record => record.kind), ['activity', 'query']); +const cases = records.map(record => checkCrossKindPollWake(record, profile)); +console.log(JSON.stringify({outcome: 'pass', cases})); diff --git a/scripts/conformance/server-parity/cross-kind-poll-contract.mjs b/scripts/conformance/server-parity/cross-kind-poll-contract.mjs new file mode 100644 index 000000000..5dceb69a9 --- /dev/null +++ b/scripts/conformance/server-parity/cross-kind-poll-contract.mjs @@ -0,0 +1,67 @@ +import assert from 'node:assert/strict'; +import {createHash} from 'node:crypto'; + +export function checkCrossKindPollWake(record, profile) { + assert.equal(record.outcome, 'observed', 'the original workflow completes'); + assert.equal(record.worker_exit, 0, 'managed worker deregisters'); + assert(['activity', 'query'].includes(record.kind)); + assert.equal(record.original_budget_seconds, 35); + assert.equal(record.cross_kind_capability, true, 'actual worker opts in'); + assert.equal(record.namespace, 'default', 'this recipe uses the explicit default namespace'); + assert.equal(record.sdk_version, profile.artifacts.sdk_php_version); + assert.equal(record.sdk_source, profile.artifacts.sdk_php_source); + assert.equal(record.adapter_sha256, profile.workload.adapter_sha256); + for (const key of ['worker_id', 'task_queue', 'workflow_id', 'run_id']) { + assert.equal(typeof record[key], 'string', key); + assert(record[key].length > 0, key); + } + for (const key of ['case_started_at', 'observed_at', 'held_observed_at', 'start_request_at', 'start_accepted_at']) { + assert(Number.isFinite(record[key]), key); + } + assert(record.observed_at <= record.case_started_at + 35, 'original budget is not rebased'); + assert(record.observed_at >= record.start_accepted_at); + assert(record.held_observed_at >= record.case_started_at); + assert(record.start_accepted_at >= record.start_request_at); + assert(record.start_request_at >= record.held_observed_at); + const heldIndex = record.polls.findIndex(poll => poll.request_id === record.held_request_id); + assert(heldIndex >= 0, 'the actual held request is retained'); + const held = record.polls[heldIndex]; + assert.equal(record.polls.filter(poll => poll.request_id === record.held_request_id).length, 1); + assert.equal(held.path, `/api/worker/${record.kind}-tasks/poll`); + assert.equal(held.requested_seconds, 5); + assert.equal(held.http, 200); + assert.equal(held.poll_status, 'task_queue_changed', 'the server emits its explicit wake hint'); + assert.equal(held.task_is_null, true, 'task is explicitly null'); + assert.equal(held.task_id, null, 'the hint grants no lease'); + assert(held.started_at <= record.held_observed_at); + assert(held.responded_at >= record.start_request_at, 'admission occurs while the request is held'); + const following = record.polls.slice(heldIndex + 1); + const sweepIndex = following.findIndex(poll => poll.path === '/api/worker/workflow-tasks/poll'); + assert(sweepIndex >= 0, 'the same worker sweeps workflow work'); + for (const poll of following.slice(0, sweepIndex + 1)) { + assert.equal(poll.requested_seconds, 0, 'no intervening held empty wait'); + assert(poll.started_at >= held.responded_at); + } + const sweep = following[sweepIndex]; + assert.equal(sweep.http, 200); + assert(['task', 'leased'].includes(sweep.poll_status)); + assert.equal(sweep.task_is_null, false); + assert.equal(typeof sweep.task_id, 'string'); + assert(sweep.task_id.length > 0, 'the ordinary poll separately grants a task'); + assert.equal(sweep.workflow_task_attempt, 1, 'the hint did not consume an earlier claim'); + assert.equal(sweep.workflow_id, record.workflow_id, 'the sweep claims the original workflow'); + assert.equal(sweep.run_id, record.run_id, 'the sweep claims the original run'); + for (const poll of [held, sweep]) { + assert(Number.isFinite(poll.started_at) && Number.isFinite(poll.responded_at)); + assert(poll.responded_at >= poll.started_at); + assert.equal(poll.worker_id, record.worker_id); + assert.equal(poll.task_queue, record.task_queue); + assert.equal(poll.namespace_header, record.namespace); + } + assert.equal(sweep.namespace_header, held.namespace_header, 'the sweep keeps its namespace'); + assert.deepEqual(record.history_events, ['StartAccepted', 'WorkflowStarted', 'ActivityScheduled', + 'ActivityStarted', 'ActivityCompleted', 'WorkflowCompleted']); + assert.equal(record.output_bytes, 1024); + assert.equal(record.output_sha256, createHash('sha256').update('s'.repeat(1024)).digest('hex')); + return {kind: record.kind, wake: true}; +} diff --git a/scripts/conformance/server-parity/cross-kind-poll-observe.php b/scripts/conformance/server-parity/cross-kind-poll-observe.php new file mode 100644 index 000000000..6b98c3bc7 --- /dev/null +++ b/scripts/conformance/server-parity/cross-kind-poll-observe.php @@ -0,0 +1,163 @@ +['pipe','r'], 1=>['pipe','w'], 2=>['file',$prefix.'.proxy.stderr','w']], $pipes, null, $environment); + if (!is_resource($proxy)) throw new RuntimeException('Proxy setup failed'); + fclose($pipes[0]); stream_set_blocking($pipes[1], false); + $worker = null; + $summary = ['kind'=>$kind, 'namespace'=>'default', 'original_budget_seconds'=>35, + 'sdk_version'=>Composer\InstalledVersions::getPrettyVersion('durable-workflow/sdk'), + 'sdk_source'=>Composer\InstalledVersions::getReference('durable-workflow/sdk'), + 'adapter_sha256'=>hash_file('sha256',$adapter)]; + $summary['case_started_at'] = microtime(true); + $deadline = $summary['case_started_at'] + 35; + try { + $url = ''; + while (!str_contains($url, "\n") && microtime(true)<$deadline) { + $url .= stream_get_contents($pipes[1]); usleep(10000); + } + $url = trim($url); + if (!preg_match('#^http://127\.0\.0\.1:[0-9]+$#', $url)) throw new RuntimeException('Proxy did not bind'); + $queue = $workflowPrefix.'-'.$kind; + $environment['DURABLE_WORKFLOW_RUNTIME_URL'] = $url; + $environment['DURABLE_WORKFLOW_TASK_QUEUE'] = $queue; + $environment['DURABLE_WORKFLOW_NAMESPACE'] = 'default'; + $environment['DURABLE_WORKFLOW_TOKEN'] = $token ?? ''; + $environment['DURABLE_WORKFLOW_CAPACITY_CELL'] = 'one-activity'; + $worker = proc_open([PHP_BINARY, __DIR__.'/cross-kind-poll-worker.php'], + [0=>['pipe','r'],1=>['file',$prefix.'.worker.stdout','w'],2=>['file',$prefix.'.worker.stderr','w']], + $workerPipes, null, $environment); + if (!is_resource($worker)) throw new RuntimeException('Worker setup failed'); + fclose($workerPipes[0]); + $held = null; + do { + $responses = array_column(crossKindPollRecords($journal), null, 'request_observation_id'); + foreach (crossKindPollRecords($arrivals) as $arrival) { + $body = json_decode($arrival['request_body'] ?? '', true); + if ($arrival['path'] === '/api/worker/'.$kind.'-tasks/poll' + && ($body['timeout_seconds'] ?? 0)>0 + && !isset($responses[$arrival['request_observation_id']])) $held = $arrival; + } + if ($held !== null) break; + if (!proc_get_status($worker)['running']) throw new RuntimeException('Worker exited before held poll'); + usleep(20000); + } while (microtime(true)<$deadline); + if ($held === null) throw new RuntimeException('Original held-poll barrier expired'); + usleep(50000); + $responses = array_column(crossKindPollRecords($journal), null, 'request_observation_id'); + if (isset($responses[$held['request_observation_id']])) throw new RuntimeException('Poll ended before admission'); + $summary['held_request_id'] = $held['request_observation_id']; + $summary['held_observed_at'] = microtime(true); + $client = new DurableWorkflow\Client($url, namespace:'default', token:$token); + $id = $workflowPrefix.'-'.$kind.'-'.bin2hex(random_bytes(8)); + $blob = str_repeat('s',1024); + $input = ['blob'=>$blob,'payload_contract'=>['workflow_input_bytes'=>1024,'workflow_result_bytes'=>1024, + 'activity_input_bytes'=>1024,'activity_result_bytes'=>1024,'signal_bytes'=>0]]; + $summary['start_request_at'] = microtime(true); + $handle = $client->startWorkflow('capacity.v1.one_activity',$id,$queue,[$input]); + $summary['start_accepted_at'] = microtime(true); + do { + $execution = $handle->describe(); + if ($execution->status === 'completed') break; + usleep(250000); + } while (microtime(true)<$deadline); + if ($execution->status !== 'completed' || $execution->output !== $blob) throw new RuntimeException('Original workflow output/deadline failed'); + $history = $client->workflowHistory($id,$handle->selectedRunId); + $events = array_column($history['events'] ?? [],'event_type'); + $expected = ['WorkflowStarted','ActivityScheduled','ActivityStarted','ActivityCompleted','WorkflowCompleted']; + if (array_values(array_intersect($events,$expected)) !== $expected) throw new RuntimeException('Semantic event order changed'); + $summary += ['workflow_id'=>$id,'run_id'=>$handle->selectedRunId,'closed_at'=>$execution->closedAt, + 'history_events'=>$events,'output_bytes'=>strlen($execution->output),'output_sha256'=>hash('sha256',$execution->output)]; + $summary['history_times'] = []; + foreach ($history['events'] as $event) $summary['history_times'][$event['event_type']] = $event['timestamp']; + $summary['observed_at'] = microtime(true); + if ($summary['observed_at'] > $deadline) throw new RuntimeException('Original observation budget expired'); + $summary['outcome'] = 'observed'; + } catch (Throwable $error) { + $summary['outcome'] = 'fail'; $summary['error'] = $error::class.': '.$error->getMessage(); + } finally { + if (is_resource($worker)) { + try { $summary['worker_exit'] = crossKindPollStopWorker($worker); } + catch(Throwable $error) { $summary['shutdown_error'] = $error->getMessage(); } + } + $summary['polls'] = []; + foreach (crossKindPollRecords($journal) as $record) { + if ($record['path'] === '/api/worker/register') { + $registration = json_decode($record['request_body'],true); + $summary['cross_kind_capability'] = in_array('cross_kind_poll_wake',$registration['capabilities']??[],true); + $summary['worker_id'] = $registration['worker_id']??null; + $summary['task_queue'] = $registration['task_queue']??null; + } + if (!str_ends_with($record['path'],'/poll')) continue; + $body = json_decode($record['request_body'],true); + $reply = json_decode($record['response_body'],true); + $summary['polls'][] = ['request_id'=>$record['request_observation_id'], 'path'=>$record['path'], + 'requested_seconds'=>$body['timeout_seconds']??null,'started_at'=>$record['request_started_at']??null, + 'responded_at'=>$record['response_observed_at']??null,'http'=>$record['status'], + 'poll_status'=>$reply['poll_status']??null,'task_id'=>$reply['task']['task_id']??null, + 'workflow_task_attempt'=>$reply['task']['workflow_task_attempt']??null, + 'workflow_id'=>$reply['task']['workflow_id']??null,'run_id'=>$reply['task']['run_id']??null, + 'task_is_null'=>is_array($reply)&&array_key_exists('task',$reply)&&$reply['task']===null, + 'worker_id'=>$body['worker_id']??null,'task_queue'=>$body['task_queue']??null, + 'namespace_header'=>$record['namespace']??null]; + } + proc_terminate($proxy, SIGTERM); fclose($pipes[1]); proc_close($proxy); + file_put_contents($prefix.'.summary.json',json_encode($summary,JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR)."\n"); + echo json_encode($summary,JSON_THROW_ON_ERROR)."\n"; + } + if (($summary['outcome']??null)!=='observed' || ($summary['worker_exit']??null)!==0) exit(1); +} diff --git a/scripts/conformance/server-parity/cross-kind-poll-worker.php b/scripts/conformance/server-parity/cross-kind-poll-worker.php new file mode 100644 index 000000000..8fe05d07f --- /dev/null +++ b/scripts/conformance/server-parity/cross-kind-poll-worker.php @@ -0,0 +1,11 @@ +run(); diff --git a/scripts/conformance/server-parity/external-consumer-check.php b/scripts/conformance/server-parity/external-consumer-check.php index d255d4f67..de942251d 100644 --- a/scripts/conformance/server-parity/external-consumer-check.php +++ b/scripts/conformance/server-parity/external-consumer-check.php @@ -80,10 +80,41 @@ function checkExternalConsumers(array $records, array $artifacts): array $require($author['encoded_bytes'] > $textBytes, "$target/$case real large payload"); $runs[$case] = $author['run_id']; } - $require(count(array_unique($runs)) === 5, "$target distinct originals"); + $completion = $data['completion-upload']; + $require(str_contains($completion['lease_owner'] ?? '', '位馃榾'), "$target actual UTF-8 completion owner"); + $require($completion['scope'] === 'raw_http_activity_completion_upload' + && $completion['parent_run_id'] === $completion['run_id'] + && $completion['task_id'] !== $completion['parent_task_id'] + && $completion['activity_completion_count'] === 1 && $completion['workflow_status'] === 'completed', + "$target completion original/task/history"); + $require($completion['result'] === [$number, null, 'range位'], "$target typed completion result"); + $require($completion['lease_expires_before'] === $completion['lease_expires_after'] + && is_string($completion['lease_expires_before']) && $completion['lease_expires_before'] !== '', + "$target upload cannot renew the lease"); + $require($completion['original_deadline'] - $completion['started_at'] === 30.0 + && $completion['finished_at'] < $completion['original_deadline'] + && $completion['elapsed_seconds'] === $completion['finished_at'] - $completion['started_at'], + "$target completion original budget"); + $require($completion['reference']['codec'] === 'avro' && $completion['reference']['size_bytes'] === 44 + && $completion['reference']['sha256'] === '72f7246e42b2f948968398c4718ce8567ad4417aee4901f8975eefd5963b9ba5', + "$target completion byte identity"); + $require(count($completion['reference']) === 5 && count($completion['duplicate_reference']) === 5, + "$target exact reference members"); + foreach (['schema', 'reference_id', 'codec', 'size_bytes', 'sha256'] as $field) { + $require(array_key_exists($field, $completion['reference']) + && array_key_exists($field, $completion['duplicate_reference']) + && $completion['reference'][$field] === $completion['duplicate_reference'][$field], + "$target duplicate reference $field identity/type"); + } + $require($completion['duplicate_key_order_changed'] === (array_keys($completion['reference']) !== array_keys($completion['duplicate_reference'])), + "$target reference member order observation"); + $require($completion['controls'] === ['identical_upload', 'conflicting_slot', 'foreign_owner', + 'closed_upload_reconciliation', 'cold_parent', 'utf8_completion_context'], "$target completion controls"); + $runs['completion-upload'] = $completion['run_id']; + $require(count(array_unique($runs)) === 6, "$target distinct originals"); $joinedRuns[$target] = $runs; } - $require(count(array_unique(array_merge(array_values($joinedRuns['php']), array_values($joinedRuns['native'])))) === 10, + $require(count(array_unique(array_merge(array_values($joinedRuns['php']), array_values($joinedRuns['native'])))) === 12, 'Implementations must have independent original runs'); foreach (['input', 'workflow-author', 'activity-author', 'arguments-1-author', 'arguments-2-author'] as $label) { foreach (['encoded_bytes', 'encoded_sha256'] as $field) { @@ -98,7 +129,7 @@ function checkExternalConsumers(array $records, array $artifacts): array $artifacts = json_decode(file_get_contents($profilePath), true, flags: JSON_THROW_ON_ERROR)['artifacts']; $labels = ['input', 'workflow-author', 'workflow-reader', 'activity-author', 'activity-reader', 'arguments-1-author', 'arguments-1-activity', 'arguments-1-reader', - 'arguments-2-author', 'arguments-2-activity', 'arguments-2-reader']; + 'arguments-2-author', 'arguments-2-activity', 'arguments-2-reader', 'completion-upload']; $records = []; foreach (['php', 'native'] as $target) { foreach ($labels as $label) { @@ -114,6 +145,11 @@ function checkExternalConsumers(array $records, array $artifacts): array 'lost_activity' => static function (array &$bad): void { $bad['native']['arguments-2-activity']['activity_completions'] = 1; }, 'changed_reference' => static function (array &$bad): void { $bad['php']['workflow-author']['reference']['sha256'] = str_repeat('0', 64); }, 'expired_observation' => static function (array &$bad): void { $bad['native']['workflow-reader']['elapsed_seconds'] = 30; }, + 'completion_other_parent' => static function (array &$bad): void { $bad['native']['completion-upload']['parent_run_id'] = 'other'; }, + 'completion_lease_renewal' => static function (array &$bad): void { $bad['php']['completion-upload']['lease_expires_after'] .= 'changed'; }, + 'completion_rounded_int64' => static function (array &$bad): void { $bad['native']['completion-upload']['result'][0] = 9007199254740992; }, + 'completion_duplicate_reference' => static function (array &$bad): void { $bad['php']['completion-upload']['duplicate_reference']['reference_id'] = 'other'; }, + 'completion_duplicate_size_type' => static function (array &$bad): void { $bad['native']['completion-upload']['duplicate_reference']['size_bytes'] = '44'; }, ]; foreach ($mutations as $label => $mutate) { $bad = $records; @@ -126,5 +162,5 @@ function checkExternalConsumers(array $records, array $artifacts): array throw new RuntimeException('Reader accepted '.$label); } echo json_encode(['outcome' => 'pass', 'scope' => 'installed_service_external_consumers', - 'original_runs' => $runs, 'scenarios_per_target' => 5, 'original_budget_seconds' => 30, + 'original_runs' => $runs, 'scenarios_per_target' => 6, 'original_budget_seconds' => 30, 'reader_refusals' => array_keys($mutations), 'embedded_transport' => 'inapplicable'], JSON_THROW_ON_ERROR)."\n"; diff --git a/scripts/conformance/server-parity/external-consumer-profile.sh b/scripts/conformance/server-parity/external-consumer-profile.sh index 5d99398f6..faf9b86b5 100644 --- a/scripts/conformance/server-parity/external-consumer-profile.sh +++ b/scripts/conformance/server-parity/external-consumer-profile.sh @@ -82,6 +82,7 @@ for target in php native; do docker run --rm --user=1000:1000 --cpus=1 --memory=512m --memory-swap=512m --network "$RESOURCE_SCOPE" \ --entrypoint php -v "$PWD:/source:ro" -v "$PWD/$evidence:/evidence" -w /source \ -e DW_PARITY_SDK_AUTOLOAD="/source/$sdk/vendor/autoload.php" \ + -e DW_PARITY_TOKEN=external-fixture-token \ -e DW_PARITY_EXTERNAL_ACTIVITY_COUNT "${image}" "scripts/conformance/server-parity/$script.php" \ "$url" "$@" > "$evidence/$target/$label.json" 2> "$evidence/$target/$label.stderr" } @@ -96,6 +97,7 @@ for target in php native; do consumer "arguments-$count-activity" external-arguments-worker "external-arguments-$count" activity "/evidence/$target/arguments-$count-author.json" consumer "arguments-$count-reader" external-arguments-worker "external-arguments-$count" reader "/evidence/$target/arguments-$count-author.json" done + consumer completion-upload activity-completion-upload-probe external-completion-upload done docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m --network=none --entrypoint php \ -v "$PWD:/source:ro" -v "$PWD/$evidence:/evidence:ro" -w /source "$image" \ diff --git a/tests/Unit/OpenApiDocumentEvolutionTest.php b/tests/Unit/OpenApiDocumentEvolutionTest.php index 1261b923b..135121125 100644 --- a/tests/Unit/OpenApiDocumentEvolutionTest.php +++ b/tests/Unit/OpenApiDocumentEvolutionTest.php @@ -3,14 +3,40 @@ namespace Tests\Unit; use DurableWorkflow\Server\Ci\OpenApiDocumentEvolution; +use PHPUnit\Framework\AssertionFailedError; use PHPUnit\Framework\TestCase; use RuntimeException; use Symfony\Component\Yaml\Yaml; +use Tests\Support\OpenApiSchema; require_once dirname(__DIR__, 2).'/scripts/ci/OpenApiDocumentEvolution.php'; class OpenApiDocumentEvolutionTest extends TestCase { + public function test_empty_poll_schema_accepts_the_published_wake_and_ordinary_empty_replies(): void + { + $schema = OpenApiSchema::fromFile(dirname(__DIR__, 2).'/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml'); + foreach (['empty', 'timeout', 'task_queue_changed'] as $status) { + $schema->assertReferenceMatches('#/components/schemas/EmptyPoll/allOf/1', (object) ['task' => null, 'poll_status' => $status]); + } + } + + public function test_empty_poll_wake_schema_does_not_grant_a_task(): void + { + $this->expectException(AssertionFailedError::class); + OpenApiSchema::fromFile(dirname(__DIR__, 2).'/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml') + ->assertReferenceMatches('#/components/schemas/EmptyPoll/allOf/1', (object) [ + 'task' => (object) ['task_id' => 'not-a-wake'], 'poll_status' => 'task_queue_changed', + ]); + } + + public function test_empty_poll_schema_preserves_unknown_status_refusal(): void + { + $this->expectException(AssertionFailedError::class); + OpenApiSchema::fromFile(dirname(__DIR__, 2).'/resources/platform-protocol-specs/worker-protocol-api.openapi.yaml') + ->assertReferenceMatches('#/components/schemas/EmptyPoll/allOf/1', (object) ['task' => null, 'poll_status' => 'unknown']); + } + public function test_candidate_structure_and_feature_ci_compare_to_the_base_revision(): void { $workflow = file_get_contents(dirname(__DIR__, 2).'/.github/workflows/phpunit-feature.yml'); diff --git a/tests/Unit/ServerParityCrossKindPollWakeTest.mjs b/tests/Unit/ServerParityCrossKindPollWakeTest.mjs new file mode 100644 index 000000000..af3a447a0 --- /dev/null +++ b/tests/Unit/ServerParityCrossKindPollWakeTest.mjs @@ -0,0 +1,60 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkCrossKindPollWake} from '../../scripts/conformance/server-parity/cross-kind-poll-contract.mjs'; + +const profile = JSON.parse(readFileSync(new URL('../../benchmarks/server-port/v2/matched.json', import.meta.url))); +function observation(kind) { + const poll = {worker_id: 'original-worker', task_queue: 'original-queue', namespace_header: 'default', http: 200}; + return { + outcome: 'observed', worker_exit: 0, kind, original_budget_seconds: 35, cross_kind_capability: true, + sdk_version: profile.artifacts.sdk_php_version, sdk_source: profile.artifacts.sdk_php_source, + adapter_sha256: profile.workload.adapter_sha256, + worker_id: 'original-worker', task_queue: 'original-queue', namespace: 'default', workflow_id: 'original-workflow', run_id: 'original-run', + case_started_at: 100, held_observed_at: 110, start_request_at: 111, start_accepted_at: 112, observed_at: 114, + held_request_id: 2, + polls: [ + {...poll, request_id: 2, path: `/api/worker/${kind}-tasks/poll`, requested_seconds: 5, + started_at: 109.9, responded_at: 111.8, poll_status: 'task_queue_changed', task_is_null: true, task_id: null}, + ...(kind === 'activity' ? [{...poll, request_id: 3, path: '/api/worker/query-tasks/poll', requested_seconds: 0, + started_at: 111.9, responded_at: 112, poll_status: 'empty', task_is_null: true, task_id: null}] : []), + {...poll, request_id: 4, path: '/api/worker/workflow-tasks/poll', requested_seconds: 0, + started_at: 112, responded_at: 112.1, poll_status: 'leased', task_is_null: false, task_id: 'original-task', workflow_task_attempt: 1, + workflow_id: 'original-workflow', run_id: 'original-run'}, + ], + history_events: ['StartAccepted', 'WorkflowStarted', 'ActivityScheduled', 'ActivityStarted', 'ActivityCompleted', 'WorkflowCompleted'], + output_bytes: 1024, output_sha256: '004ff2ad28c856e1e1eccdb7a0222f100eb2c26b6301a9f5968cdda899307987', + }; +} +for (const kind of ['activity', 'query']) test(`${kind} hint and separate workflow claim are required`, () => { + assert.deepEqual(checkCrossKindPollWake(observation(kind), profile), {kind, wake: true}); +}); +const corruptions = { + 'completed work after a full empty wait': r => { r.polls[0].poll_status = 'timeout'; }, + 'empty response without a wake': r => { r.polls[0].poll_status = 'empty'; }, + 'no capability opt-in': r => { r.cross_kind_capability = false; }, + 'hint claims a task': r => { r.polls[0].task_id = 'illicit-lease'; }, + 'missing explicit null task': r => { delete r.polls[0].task_is_null; }, + 'hint belongs to another queue': r => { r.polls[0].task_queue = 'foreign-queue'; }, + 'hint belongs to another worker': r => { r.polls[0].worker_id = 'foreign-worker'; }, + 'sweep changes namespace': r => { r.polls.at(-1).namespace_header = 'foreign'; }, + 'both polls use a foreign namespace': r => { for (const poll of r.polls) poll.namespace_header = 'foreign'; }, + 'sweep waits again': r => { r.polls.at(-1).requested_seconds = 5; }, + 'sweep grants no task': r => { r.polls.at(-1).task_id = null; }, + 'hint consumed an earlier claim': r => { r.polls.at(-1).workflow_task_attempt = 2; }, + 'sweep claims another workflow': r => { r.polls.at(-1).workflow_id = 'different'; }, + 'sweep claims another run': r => { r.polls.at(-1).run_id = 'different'; }, + 'hint ended before admission': r => { r.polls[0].responded_at = 110.5; }, + 'sweep precedes wake': r => { r.polls.at(-1).started_at = 110.5; }, + 'wrong held request': r => { r.held_request_id = 99; }, + 'HTTP refusal': r => { r.polls[0].http = 503; }, + 'later rebudget': r => { r.observed_at = 136; }, + 'changed durable history': r => { r.history_events.push('WorkflowCompleted'); }, + 'wrong exact output': r => { r.output_sha256 = 'changed'; }, + 'different installed SDK': r => { r.sdk_source = 'different'; }, +}; +for (const [name, corrupt] of Object.entries(corruptions)) test(`cross-kind reader refuses ${name}`, () => { + const record = observation('query'); + corrupt(record); + assert.throws(() => checkCrossKindPollWake(record, profile)); +}); diff --git a/tests/Unit/ServerParityRunnerTest.mjs b/tests/Unit/ServerParityRunnerTest.mjs index 1f3a7fe80..4c1629f60 100644 --- a/tests/Unit/ServerParityRunnerTest.mjs +++ b/tests/Unit/ServerParityRunnerTest.mjs @@ -3,6 +3,7 @@ import assert from 'node:assert/strict'; import {readFileSync} from 'node:fs'; import {checkObservation, compareRecords as compareCorpusRecords} from '../../scripts/conformance/server-parity/contract.mjs'; import {parseDatabaseResourceLimits} from '../../scripts/conformance/server-parity/database-resource-contract.mjs'; +import './ServerParityCrossKindPollWakeTest.mjs'; const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/one-activity.json', import.meta.url))); const reviewed = {'one-activity.json': fixture};