From 332f5026d062a551b72fd54fc65cdd24122cc989 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 15:05:57 +0000 Subject: [PATCH] Wait DW_WORKER_POLL_TIMEOUT on polls and check operation names like PHP A workflow or activity poll without timeout_seconds now waits max(0, DW_WORKER_POLL_TIMEOUT) seconds (30 by default), as Server 2.10.0's LongPoller does, instead of 5. A query-task poll starts from the same setting, capped by DW_QUERY_TASK_POLL_TIMEOUT. An explicit null wait takes the default, and a requested wait follows Laravel's integer rule from 0 to 60. Signal, update and query names get PHP's validateOperationName errors: empty, over 256 bytes or with control characters. Native keeps its 255-byte storage guard for signal and update names, which PHP's own columns cannot exceed. A query poll from a worker registered without capabilities answered 503; it now polls empty, as PHP does. --- .../2026-10-22-poll-waits.md | 42 +++++++ docs/rust-server-port.md | 10 +- rust/src/runtime/execution.rs | 14 ++- rust/src/runtime/http.rs | 6 +- rust/src/runtime/laravel_validation.rs | 17 +++ rust/src/runtime/queries.rs | 17 ++- rust/src/runtime/query_authority.rs | 5 + rust/src/runtime/store.rs | 23 +++- rust/src/runtime/updates.rs | 2 + rust/tests/execution.rs | 2 + rust/tests/execution/poll_waits.rs | 118 ++++++++++++++++++ 11 files changed, 236 insertions(+), 20 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-22-poll-waits.md create mode 100644 rust/tests/execution/poll_waits.rs diff --git a/docs/rust-server-port-checkpoints/2026-10-22-poll-waits.md b/docs/rust-server-port-checkpoints/2026-10-22-poll-waits.md new file mode 100644 index 000000000..bb1c9ad37 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-22-poll-waits.md @@ -0,0 +1,42 @@ +# October 22, worker poll waits and operation names + +- **Gap.** A workflow or activity poll without `timeout_seconds` waited 5 + seconds; PHP waits `DW_WORKER_POLL_TIMEOUT` seconds, 30 by default. A + query-task poll without one also waited 5 seconds where PHP starts from + the same setting. Native refused an explicit `null` wait, which PHP's + `nullable` rule accepts. +- **Source.** Server 2.10.0's `LongPoller::until`, + `WorkflowQueryTaskBroker::queryPollTimeoutSeconds` and + `WorkflowController::validateOperationName`. +- **Poll waits.** + - A workflow or activity poll without a wait, or with `null`, waits + `max(0, DW_WORKER_POLL_TIMEOUT)` seconds. The setting is not capped. + - A query-task poll waits the requested time or that setting, capped by + `DW_QUERY_TASK_POLL_TIMEOUT` (default 5); zero answers at once. + - A requested wait passes Laravel's `integer` rule (so `"5"` is five + seconds) and must lie from 0 to 60. +- **Operation names.** + - Signal, update and query requests check the name first, after the + payload transport check, as PHP's controller does. Empty names, names + over 256 bytes and names with ASCII control characters get PHP's + `validation_failed` errors and messages. + - PHP's signal and update name columns hold 255 bytes. Native keeps its + storage guard and refuses a 256-byte or all-space name with + `invalid_signal_name` or `invalid_update_name`. Query names are not + stored in those columns, so a 256-byte query name is accepted. +- **Fixed along the way.** A query poll from a worker registered without + `capabilities` answered 503 `storage_unavailable`, because the claim read + the missing list as required. It now polls empty, as PHP's `stringArray` + does. +- **Tests** (`poll_waits.rs`). + - Workflow, activity and query polls without a wait, or with `null`, take + the configured second. + - A zero setting and a requested `"0"` answer at once. + - Requested waits of 61, -1, `"soon"` and 1.5 are refused. + - A task started during a default wait is claimed. + - A 255-byte signal name is accepted; 256-byte signal and update names + are refused with the storage reasons. Too-long names and names with + control characters get PHP's messages for each kind. + - Full SQLite run (all targets): 228 unit, 397 execution and 2 upgrade + tests passed. Poll, signal, update and query tests: PostgreSQL 122, + MySQL 122. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index bd7b02079..81eb05390 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -122,8 +122,14 @@ contract fixture before its correction. non-default value of the other 67 ([checkpoint](rust-server-port-checkpoints/2026-10-22-environment-contract.md)). Several of those are cheap to honor: repair timing and the multi-node - declaration. Native's defaults still differ from PHP's for two honored - settings: the worker poll timeout and operation name length. + declaration. +- **Signal and update names PHP cannot store.** Native validates signal, + update and query names as PHP does (1 to 256 bytes, no control + characters). PHP's signal and update name columns hold 255 bytes, so PHP + stores a 256-byte name on SQLite and fails with a database error on MySQL + and PostgreSQL. Native refuses such a name, and a name of only spaces, + with `invalid_signal_name` or `invalid_update_name` on every database + ([checkpoint](rust-server-port-checkpoints/2026-10-22-poll-waits.md)). - **Key order inside plain JSON input.** For input that is not an envelope, PHP encodes nested objects as Avro maps in request order; native encodes them with sorted keys. The decoded values are equal, but the blobs differ. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 7fd36531f..82d166673 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -149,6 +149,11 @@ impl Runtime { max: self.queries.history_page_size_max, } } + /// The wait of a workflow or activity poll that names none + /// (DW_WORKER_POLL_TIMEOUT). + pub(super) fn worker_poll_wait(&self) -> u64 { + u64::try_from(self.queries.worker_poll_timeout_seconds).unwrap_or(0) + } /// Immutable configured-token cohort. Database transitions and scheduler /// stay shared; configuring credentials does not mutate process state. pub fn with_role_tokens( @@ -513,6 +518,7 @@ impl Runtime { ) -> Result { let body = super::start_request::normalized(body); super::input_envelope::transport(&body)?; + super::laravel_validation::operation_name(name, "signal")?; super::command_requests::signal(&body)?; let arguments = self.command_arguments(&body).await?; let prepared = super::store::prepare_control_arguments( @@ -629,6 +635,7 @@ impl Runtime { ) -> Result<(StatusCode, Value)> { let body = super::start_request::normalized(body); super::input_envelope::transport(&body)?; + super::laravel_validation::operation_name(name, "update")?; super::command_requests::update(&body)?; let arguments = self.command_arguments(&body).await?; if arguments.len() > super::queries::ARGUMENT_BYTES { @@ -684,9 +691,7 @@ impl Runtime { } error }; - if name.len() > 256 { - return Err(invalid("query_name", "The query name exceeds the maximum length of 256 bytes.")); - } + super::laravel_validation::operation_name(name, "query")?; self.query_target(workflow_id, run_id).await?; let input = super::laravel_validation::normalized(body.get("input")).filter(|value| !value.is_null()); if input.as_ref().is_some_and(|value| !value.is_object() && !value.is_array()) { @@ -945,8 +950,9 @@ impl Runtime { Storage::MySql(store) => store.deregister_fenced(worker_id, body, codec).await, } } + #[cfg(test)] delegate!(poll(body: Value, kind: &'static str, protocol_version: &str) -> Result); - delegate!(poll_with_pages(body: Value, kind: &'static str, protocol_version: &str, pages: super::history_pages::HistoryPages) -> Result); + delegate!(poll_with_pages(body: Value, kind: &'static str, protocol_version: &str, pages: super::history_pages::HistoryPages, default_wait: u64) -> Result); delegate!(heartbeat_task(task_id: &str, body: Value) -> Result); pub(crate) async fn deliver_cancellation(&self, task_id: &str, body: Value, protocol: &str) -> Result { let codec = self.transition_codec.clone(); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index b6bb73eef..fd6dea929 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -1539,8 +1539,9 @@ async fn poll_workflow( Json(body): Json, ) -> Result> { let pages = runtime.history_pages(); + let wait = runtime.worker_poll_wait(); runtime - .poll_with_pages(body, "workflow", request_protocol(&headers)?, pages) + .poll_with_pages(body, "workflow", request_protocol(&headers)?, pages, wait) .await .map(Json) } @@ -1549,8 +1550,9 @@ async fn poll_activity( headers: HeaderMap, Json(body): Json, ) -> Result> { + let wait = runtime.worker_poll_wait(); runtime - .poll(body, "activity", request_protocol(&headers)?) + .poll_with_pages(body, "activity", request_protocol(&headers)?, Default::default(), wait) .await .map(Json) } diff --git a/rust/src/runtime/laravel_validation.rs b/rust/src/runtime/laravel_validation.rs index f3623fad6..064aaa6e7 100644 --- a/rust/src/runtime/laravel_validation.rs +++ b/rust/src/runtime/laravel_validation.rs @@ -248,3 +248,20 @@ fn messages( } found } + +/// `WorkflowController::validateOperationName`, which a signal, update or +/// query request runs first: the name is one to 256 bytes +/// (DW_MAX_OPERATION_NAME_LENGTH, which native runs only at its default) +/// without ASCII control characters. +pub(super) fn operation_name(name: &str, kind: &str) -> super::Result<()> { + let message = if name.is_empty() { + format!("The {kind} name must not be empty.") + } else if name.len() > 256 { + format!("The {kind} name exceeds the maximum length of 256 bytes.") + } else if name.bytes().any(|byte| byte < 0x20 || byte == 0x7f) { + format!("The {kind} name must not contain control characters.") + } else { + return Ok(()); + }; + Err(failure_messages(vec![(format!("{kind}_name"), vec![message])]).expect("one message")) +} diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index b2245e017..eee15364b 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -553,11 +553,15 @@ where pub(super) async fn poll_query(&self, body: Value, config: &QueryTaskConfig) -> Result { let owner = text(&body, "worker_id")?; text(&body, "task_queue")?; - let timeout = body - .get("timeout_seconds") - .map_or(Some(5), Value::as_u64) - .filter(|v| *v <= 60) - .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?; + // `queryPollTimeoutSeconds`: the requested wait, or the worker poll + // wait (DW_WORKER_POLL_TIMEOUT), capped by DW_QUERY_TASK_POLL_TIMEOUT. + let timeout = match body.get("timeout_seconds") { + None | Some(Value::Null) => u64::try_from(config.worker_poll_timeout_seconds).unwrap_or(0), + Some(value) => super::laravel_validation::integer(value) + .filter(|v| (0..=60).contains(v)) + .and_then(|v| u64::try_from(v).ok()) + .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?, + }; let timeout = timeout.min(config.poll_timeout_seconds.max(1) as u64); let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); // One request belongs to the registration observed before its wait. @@ -788,7 +792,8 @@ where return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch")); } let types = DB::document_row(&worker, "supported_workflow_types")?; - let capabilities = DB::document_row(&worker, "capabilities")?; + // A registration without capabilities has none (`stringArray`). + let capabilities = DB::optional_document_row(&worker, "capabilities")?.unwrap_or_default(); if !capabilities .as_array() .is_some_and(|values| values.iter().any(|v| v == "query_tasks")) diff --git a/rust/src/runtime/query_authority.rs b/rust/src/runtime/query_authority.rs index f23b5315f..11b0bffeb 100644 --- a/rust/src/runtime/query_authority.rs +++ b/rust/src/runtime/query_authority.rs @@ -27,6 +27,9 @@ pub struct QueryTaskConfig { pub history_page_size_default: i64, pub history_page_size_max: i64, pub poll_timeout_seconds: i64, + /// How long a workflow or activity poll that names no `timeout_seconds` + /// waits, as PHP's `LongPoller`: `max(0, DW_WORKER_POLL_TIMEOUT)`. + pub worker_poll_timeout_seconds: i64, } impl Default for QueryTaskConfig { @@ -43,6 +46,7 @@ impl Default for QueryTaskConfig { history_page_size_default: 500, history_page_size_max: 1000, poll_timeout_seconds: 5, + worker_poll_timeout_seconds: 30, } } } @@ -118,6 +122,7 @@ impl QueryTaskConfig { 5, )? .max(1), + worker_poll_timeout_seconds: poll.max(0), }) } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 5f1a66539..19a4f9168 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -567,6 +567,8 @@ where // The authenticated request's context; a direct call records only // the request id and signal name. let context = context.unwrap_or_else(|| json!({"request":{"request_id":body["request_id"]},"signal_name":name})); + // PHP's signal_name column holds 255 bytes, one fewer than its + // validation allows; native refuses what it cannot store. if name.trim().is_empty() || name.len() > 255 { return Err(refuse( StatusCode::UNPROCESSABLE_ENTITY, @@ -1252,31 +1254,40 @@ where ) } + /// A poll with the default page sizes and PHP's default 30-second wait. + #[cfg(test)] pub(crate) async fn poll( &self, body: Value, kind: &'static str, protocol_version: &str, ) -> Result { - self.poll_with_pages(body, kind, protocol_version, Default::default()) + self.poll_with_pages(body, kind, protocol_version, Default::default(), 30) .await } /// A workflow or activity poll, with the configured history page sizes. + /// A poll that names no `timeout_seconds` waits `default_wait` seconds, + /// as PHP's `LongPoller` waits DW_WORKER_POLL_TIMEOUT. pub(crate) async fn poll_with_pages( &self, body: Value, kind: &'static str, protocol_version: &str, pages: super::history_pages::HistoryPages, + default_wait: u64, ) -> Result { text(&body, "worker_id")?; text(&body, "task_queue")?; - let timeout = body - .get("timeout_seconds") - .map_or(Some(5), Value::as_u64) - .filter(|t| *t <= 60) - .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?; + // Laravel's `nullable|integer|min:0|max:60` bounds a requested wait; + // the configured default is not capped. + let timeout = match body.get("timeout_seconds") { + None | Some(Value::Null) => default_wait, + Some(value) => super::laravel_validation::integer(value) + .filter(|t| (0..=60).contains(t)) + .and_then(|t| u64::try_from(t).ok()) + .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?, + }; let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); let requested = super::history_pages::requested_size(&body)?; let encoding = if kind == "workflow" { diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 6bbdb5f10..f92870ba2 100644 --- a/rust/src/runtime/updates.rs +++ b/rust/src/runtime/updates.rs @@ -304,6 +304,8 @@ where context: Option, codec: Arc, ) -> Result<(StatusCode, Value)> { + // PHP's update_name column holds 255 bytes, one fewer than its + // validation allows; native refuses what it cannot store. if name.trim().is_empty() || name.len() > 255 { return Err(refuse( StatusCode::UNPROCESSABLE_ENTITY, diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 746707a9a..a0961da09 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -103,6 +103,8 @@ mod parent_close_policies; mod history_pages; #[path = "execution/declared_signals.rs"] mod declared_signals; +#[path = "execution/poll_waits.rs"] +mod poll_waits; #[path = "execution/child_cascade.rs"] mod child_cascade; #[path = "execution/completion_open_work.rs"] diff --git a/rust/tests/execution/poll_waits.rs b/rust/tests/execution/poll_waits.rs new file mode 100644 index 000000000..89a8c06db --- /dev/null +++ b/rust/tests/execution/poll_waits.rs @@ -0,0 +1,118 @@ +use super::*; +use durable_workflow_server::runtime::QueryTaskConfig; +use std::time::{Duration, Instant}; + +// Server 2.10.0's `LongPoller`: a worker poll that names no `timeout_seconds` +// (or names null) waits DW_WORKER_POLL_TIMEOUT seconds, 30 by default. A +// requested wait is a Laravel integer from 0 to 60; the setting itself is not +// capped. A query-task poll caps either wait by DW_QUERY_TASK_POLL_TIMEOUT. + +async fn timed_poll(app: &Router, path: &str, body: Value) -> (StatusCode, Value, Duration) { + let began = Instant::now(); + let (status, response) = request(app, "POST", path, body).await; + (status, response, began.elapsed()) +} + +const WORKFLOW: &str = "/api/worker/workflow-tasks/poll"; +const ACTIVITY: &str = "/api/worker/activity-tasks/poll"; +const QUERY: &str = "/api/worker/query-tasks/poll"; + +#[tokio::test] +async fn polls_without_a_wait_use_the_configured_worker_poll_wait() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let configured = |seconds| QueryTaskConfig { worker_poll_timeout_seconds: seconds, ..Default::default() }; + let one = router(runtime.clone().with_query_task_config(configured(1))); + let zero = router(runtime.with_query_task_config(configured(0))); + register(&one, "worker", json!(["echo"]), json!(["greet"])).await; + + // One configured second, where native used to wait five. + for (path, body) in [ + (WORKFLOW, json!({"worker_id":"worker","task_queue":"test"})), + (WORKFLOW, json!({"worker_id":"worker","task_queue":"test","timeout_seconds":null})), + (ACTIVITY, json!({"worker_id":"worker","task_queue":"test"})), + (QUERY, json!({"worker_id":"worker","task_queue":"test"})), + ] { + let (status, response, waited) = timed_poll(&one, path, body.clone()).await; + assert_eq!(status, StatusCode::OK, "{path} {body}: {response}"); + assert!(response["task"].is_null(), "{response}"); + assert!(waited >= Duration::from_millis(900) && waited < Duration::from_secs(4), "{path} {body} waited {waited:?}"); + } + + // A zero setting answers at once, as does a requested zero given as a + // string. + for (app, body) in [ + (&zero, json!({"worker_id":"worker","task_queue":"test"})), + (&one, json!({"worker_id":"worker","task_queue":"test","timeout_seconds":"0"})), + ] { + let (status, response, waited) = timed_poll(app, WORKFLOW, body.clone()).await; + assert_eq!(status, StatusCode::OK, "{body}: {response}"); + assert!(waited < Duration::from_millis(800), "{body} waited {waited:?}"); + } + + // A requested wait outside 0 through 60 is refused. + for timeout in [json!(61), json!(-1), json!("soon"), json!(1.5)] { + for path in [WORKFLOW, ACTIVITY, QUERY] { + let (status, response) = + request(&one, "POST", path, json!({"worker_id":"worker","task_queue":"test","timeout_seconds":timeout})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{path} {timeout}: {response}"); + } + } +} + +#[tokio::test] +async fn a_task_that_arrives_during_the_default_wait_is_claimed() { + let database = TestDatabase::new().await; + let runtime = database + .open() + .await + .unwrap() + .with_query_task_config(QueryTaskConfig { worker_poll_timeout_seconds: 10, ..Default::default() }); + let app = router(runtime); + register(&app, "worker", json!(["echo"]), json!([])).await; + let waiting = { + let app = app.clone(); + tokio::spawn(async move { timed_poll(&app, WORKFLOW, json!({"worker_id":"worker","task_queue":"test"})).await }) + }; + tokio::time::sleep(Duration::from_millis(300)).await; + let started = start(&app, "arrives-while-waiting").await; + let (status, response, waited) = waiting.await.unwrap(); + assert_eq!(status, StatusCode::OK, "{response}"); + assert_eq!(response["task"]["run_id"], started["run_id"], "{response}"); + assert!(waited < Duration::from_secs(8), "waited {waited:?}"); +} + +// `WorkflowController::validateOperationName` runs first on a signal, update +// or query: a name is one to 256 bytes without control characters. +#[tokio::test] +async fn operation_names_are_validated_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + start(&app, "named").await; + // PHP stores signal and update names in 255-byte columns, so native + // refuses a 256-byte name its validation would pass. + let (status, body) = request(&app, "POST", &format!("/api/workflows/named/signal/{}", "s".repeat(255)), json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "a 255-byte signal name is accepted: {body}"); + for (kind, reason) in [("signal", "invalid_signal_name"), ("update", "invalid_update_name")] { + let (status, body) = request(&app, "POST", &format!("/api/workflows/named/{kind}/{}", "s".repeat(256)), json!({})).await; + assert_eq!((status, body["reason"].as_str()), (StatusCode::UNPROCESSABLE_ENTITY, Some(reason)), "{body}"); + } + let (status, body) = request(&app, "POST", &format!("/api/workflows/missing/query/{}", "q".repeat(256)), json!({})).await; + assert_eq!((status, body["reason"].as_str()), (StatusCode::NOT_FOUND, Some("instance_not_found")), "a 256-byte query name passes validation: {body}"); + + let too_long = "s".repeat(257); + for (kind, name, message) in [ + ("signal", too_long.as_str(), "The signal name exceeds the maximum length of 256 bytes."), + ("signal", "bad%0Aname", "The signal name must not contain control characters."), + ("update", too_long.as_str(), "The update name exceeds the maximum length of 256 bytes."), + ("update", "bad%7Fname", "The update name must not contain control characters."), + ("query", too_long.as_str(), "The query name exceeds the maximum length of 256 bytes."), + ("query", "bad%09name", "The query name must not contain control characters."), + ] { + let (status, body) = request(&app, "POST", &format!("/api/workflows/named/{kind}/{name}"), json!({})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{kind} {name}: {body}"); + assert_eq!(body["reason"], "validation_failed", "{body}"); + assert_eq!(body["errors"][format!("{kind}_name")], json!([message]), "{body}"); + assert_eq!(body["workflow_id"], "named", "{body}"); + } +}