From f1a5d876fbbb8acdb74703f5a62a2e484c3bbe50 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:33:56 +0000 Subject: [PATCH 1/4] Honor negotiated workflow completion request limits --- CHANGELOG.md | 9 ++ README.md | 19 ++- src/lib.rs | 1 + src/runtime_uploads.rs | 57 ++++++- src/tests/completion_limits.rs | 288 +++++++++++++++++++++++++++++++++ 5 files changed, 366 insertions(+), 8 deletions(-) create mode 100644 src/tests/completion_limits.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index f2566922..5041b1d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +- Completion payload planning reads the optional advertised + `limits.max_workflow_task_completion_bytes` separately from ordinary request + and per-value limits. Individually valid inline batches no longer demand + unavailable external storage when the Server advertises this allowance and + storage is explicitly unconfigured, disabled or unsupported. Missing fields + retain the conservative ordinary cap; malformed fields refuse completion + planning before uploads. Server #1136 publication and installed fan-out and + over-limit qualification remain required for delivery of SDK #254. + ## 4.15.0 - All workflow and run routes encode each identifier or name as one path diff --git a/README.md b/README.md index 2efb8170..4beaeb73 100644 --- a/README.md +++ b/README.md @@ -724,10 +724,21 @@ payloads; if those fixed inputs cannot fit, reporting refuses explicitly. ### Workflow task completion size A workflow task's commands travel to the Server in one completion request. -Native Server 3.6.0 and later accept a completion up to 16 times the -per-payload limit (`DW_MAX_PAYLOAD_BYTES`, 2 MiB by default, so 32 MiB), while -each command's arguments must still fit the per-payload limit. A larger -completion, usually a very large fan-out, is refused with HTTP 413, +The SDK reads the optional `limits.max_workflow_task_completion_bytes` from +Server discovery for the completion's encoded JSON request size. Each command's +arguments still obey the per-payload limit. An explicit positive aggregate cap +allows individually valid values to remain in one inline batch when namespace +payload storage is unconfigured, disabled or unsupported. Configured storage +retains its upload and lease checks. + +Without that discovery field, the planner conservatively uses the ordinary +request cap. It can require external storage for an aggregate above that cap +even if the Server would accept the batch inline. The field's publication and +installed fan-out qualification are tracked in +[Server #1136](https://github.com/durable-workflow/server/issues/1136) and +[SDK #254](https://github.com/durable-workflow/sdk-rust/issues/254). + +A completion above the Server's aggregate allowance is refused with HTTP 413, `reason: "payload_too_large"` and `limit_kind: "workflow_task_completion_bytes"`. Automatic structural failure requires the HTTP 413 to reach the worker. A diff --git a/src/lib.rs b/src/lib.rs index 61f9ae1b..3e0cf885 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19745,6 +19745,7 @@ mod tests { mod cancellation_scope_opening; mod cancellation_scope_replay; mod child_workflow_policies; + mod completion_limits; mod condition_fingerprint; mod cooperative_cancellation; mod local_activity; diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs index 0f320640..4000a528 100644 --- a/src/runtime_uploads.rs +++ b/src/runtime_uploads.rs @@ -10,6 +10,8 @@ pub(super) struct Policy { threshold: usize, max_bytes: usize, request_bytes: usize, + completion_request_bytes: std::result::Result, ()>, + inline_completion: bool, timeout: Duration, available: bool, completion_context: bool, @@ -32,13 +34,23 @@ impl Policy { } else { 2 * 1024 * 1024 }; + let completion_request_bytes = info + .pointer("/limits/max_workflow_task_completion_bytes") + .map(|limit| positive(limit).ok_or(())) + .transpose(); let storage = &info["namespace"]["external_payload_storage"]; + let inline_completion = matches!( + storage["status"].as_str(), + Some("unconfigured" | "disabled" | "unsupported") + ); let manifest = &storage["transport"]; if manifest.is_null() { return Ok(Self { threshold: request_bytes, max_bytes: request_bytes, request_bytes, + completion_request_bytes, + inline_completion, timeout: Duration::from_secs(30), available: false, completion_context: false, @@ -66,6 +78,8 @@ impl Policy { threshold, max_bytes, request_bytes, + completion_request_bytes, + inline_completion, timeout: Duration::from_secs(timeout as u64), available: storage["status"] == "available", completion_context: manifest["upload"]["completion_context"]["schema"] @@ -158,6 +172,14 @@ impl Client { return Ok(()); } let policy = self.runtime_upload_policy(protocol).await?; + let completion_bytes = if is_command_completion(path, protocol) { + policy + .completion_request_bytes + .map_err(|()| unsupported("invalid workflow completion request limit"))? + } else { + None + }; + let request_bytes = completion_bytes.unwrap_or(policy.request_bytes); // Plan every replacement before any upload. Aggregate JSON size matters, // even when each individual value is below the namespace threshold. payloads.sort_by_key(|(_, size)| std::cmp::Reverse(*size)); @@ -166,6 +188,18 @@ impl Client { "external_payload_oversized: encoded payload exceeds runtime upload limit".into(), )); } + // A producer-advertised aggregate cap does not require a payload store. + // Preserve one inline batch when every value already meets admission. + // The Server owns an over-cap refusal, including its actual HTTP status. + if completion_bytes.is_some() + && policy.inline_completion + && !policy.available + && payloads + .iter() + .all(|(_, size)| *size <= policy.threshold && *size <= policy.request_bytes) + { + return Ok(()); + } let mut uploads = Vec::new(); for (path, size) in &payloads { if *size > policy.threshold { @@ -173,15 +207,15 @@ impl Client { } } for (path, _) in &payloads { - if serde_json::to_vec(body)?.len() <= policy.request_bytes { + if serde_json::to_vec(body)?.len() <= request_bytes { break; } if !uploads.iter().any(|upload| upload.path == *path) { uploads.push(select(body, path)?); } } - if serde_json::to_vec(body)?.len() > policy.request_bytes { - return Err(Error::Codec("payload_too_large: request metadata exceeds ordinary API limit with payloads externalized".into())); + if serde_json::to_vec(body)?.len() > request_bytes { + return Err(Error::Codec("payload_too_large: request metadata exceeds advertised request limit with payloads externalized".into())); } if !uploads.is_empty() && !policy.available { return Err(Error::Codec( @@ -507,6 +541,20 @@ fn replace(body: &mut Value, path: &str, reference: Value) -> Result<()> { Ok(()) } +fn is_command_completion(path: &str, protocol: RequestProtocol) -> bool { + if !matches!( + protocol, + RequestProtocol::DefaultWorker | RequestProtocol::Worker(_) + ) { + return false; + } + let path = path.split('?').next().unwrap_or(path); + let parts: Vec<_> = path.strip_prefix('/').unwrap_or(path).split('/').collect(); + matches!(parts.as_slice(), + ["worker", "workflow-tasks", task, "complete"] + | ["worker", "activity-tasks", task, "continuation", "complete"] if !task.is_empty()) +} + // Only protocol-owned slots are eligible; arbitrary application maps are not envelopes. fn payload_paths(body: &Value, path: &str, protocol: RequestProtocol) -> Vec { let path = path.split('?').next().unwrap_or(path); @@ -517,7 +565,8 @@ fn payload_paths(body: &Value, path: &str, protocol: RequestProtocol) -> Vec { + ["worker", "workflow-tasks", _, "complete"] + | ["worker", "activity-tasks", _, "continuation", "complete"] => { for (index, command) in body["commands"] .as_array() .into_iter() diff --git a/src/tests/completion_limits.rs b/src/tests/completion_limits.rs new file mode 100644 index 00000000..9fc6ac7d --- /dev/null +++ b/src/tests/completion_limits.rs @@ -0,0 +1,288 @@ +use super::runtime_uploads::{policy, reference}; +use super::*; + +const WORKFLOW: &str = "/worker/workflow-tasks/test/complete"; +const CONTINUATION: &str = "/worker/activity-tasks/test/continuation/complete"; + +fn responses(path: &str, body: &str, _number: usize) -> Option<(&'static str, String)> { + let profile = path.split('/').nth(1).unwrap_or_default(); + if path.ends_with("/api/cluster/info") { + let mut info = policy(); + let limit = match profile { + "null" => Value::Null, + "zero" => json!(0), + "negative" => json!(-1), + "fraction" => json!(8192.5), + "string" => json!("8192"), + "boolean" => json!(true), + "list" => json!([8192]), + "object" => json!({"bytes":8192}), + "small" => json!(1024), + _ => json!(8192), + }; + if profile != "legacy" { + info["limits"]["max_workflow_task_completion_bytes"] = limit; + } + let storage = &mut info["namespace"]["external_payload_storage"]; + storage["threshold_bytes"] = json!(2048); + match profile { + "available" | "small" => {} + "transient" => storage["status"] = json!("unavailable"), + _ => { + storage["status"] = json!(match profile { + "disabled" => "disabled", + "unsupported" => "unsupported", + _ => "unconfigured", + }); + storage["transport"] = Value::Null; + } + } + return Some(("200 OK", info.to_string())); + } + if path.ends_with("/api/external-payloads/v1") { + return Some(( + "201 Created", + json!({ + "schema":"durable-workflow.v2.runtime-external-payload-upload.v1", + "transport_version":1, "reference":reference(body) + }) + .to_string(), + )); + } + if profile == "refused" && body.contains("schedule_activity") { + return Some(( + "413 Payload Too Large", + json!({ + "reason":"payload_too_large", "limit_kind":"workflow_task_completion_bytes", + "limit":8192, "message":"completion exceeds fixture cap" + }) + .to_string(), + )); + } + Some(( + "200 OK", + json!({"outcome":"completed","recorded":true}).to_string(), + )) +} + +fn setup(profile: &str) -> (MockWorkerServer, Client) { + let server = MockWorkerServer::start_with_behavior(MockWorkerBehavior { + request_override: Some(responses), + ..MockWorkerBehavior::default() + }); + let client = Client::new(format!("{}/{profile}", server.base_url())).unwrap(); + (server, client) +} + +fn batch(count: usize) -> Value { + let arguments = encode_value_envelope(&json!(["x".repeat(1000)]), DEFAULT_CODEC).unwrap(); + json!({"lease_owner":"worker", "workflow_task_attempt":2, "commands":(0..count).map(|index| { + json!({"type":"schedule_activity", "activity_type":"fixture.bytes", "activity_id":index.to_string(), + "arguments":arguments}) + }).collect::>()}) +} + +async fn send(client: &Client, path: &str, worker: bool, body: &Value) -> Result { + client + .request_json( + reqwest::Method::POST, + path, + if worker { + RequestProtocol::Worker(WORKER_PROTOCOL_VERSION) + } else { + RequestProtocol::ControlPlane + }, + Some(body), + ) + .await +} + +#[tokio::test] +async fn completion_limits_preserve_inline_atomic_batches_without_storage() { + let body = batch(3); + assert!(body.to_string().len() > 2048 && body.to_string().len() < 8192); + for profile in ["unconfigured", "disabled", "unsupported"] { + for path in [WORKFLOW, CONTINUATION] { + let (server, client) = setup(profile); + send(&client, path, true, &body).await.unwrap(); + assert_eq!( + server.request_bodies(&format!("/{profile}/api{path}")), + [body.clone()] + ); + assert_eq!( + server.request_count(&format!("/{profile}/api/external-payloads/v1")), + 0 + ); + } + } +} + +#[tokio::test] +async fn completion_limits_preserve_old_producer_and_individual_value_limits() { + let (server, client) = setup("legacy"); + let error = send(&client, WORKFLOW, true, &batch(3)).await.unwrap_err(); + assert!( + error.to_string().contains("external_payload_unavailable"), + "{error}" + ); + assert_eq!(server.captured_paths(), ["/legacy/api/cluster/info"]); + + let (server, client) = setup("unconfigured"); + let oversized = encode_value_envelope(&json!("x".repeat(3000)), DEFAULT_CODEC).unwrap(); + let body = json!({"commands":[{"type":"schedule_activity","arguments":oversized}]}); + let error = send(&client, WORKFLOW, true, &body).await.unwrap_err(); + assert!( + error.to_string().contains("external_payload_oversized"), + "{error}" + ); + assert_eq!(server.captured_paths(), ["/unconfigured/api/cluster/info"]); +} + +#[tokio::test] +async fn completion_limits_validate_explicit_values_only_on_completion_routes() { + for profile in [ + "null", "zero", "negative", "fraction", "string", "boolean", "list", "object", + ] { + for path in [WORKFLOW, CONTINUATION] { + let (server, client) = setup(profile); + let error = send(&client, path, true, &batch(3)).await.unwrap_err(); + assert!( + error.to_string().contains( + "external_payload_unsupported: invalid workflow completion request limit" + ), + "{profile} {path}: {error}" + ); + assert_eq!( + server.captured_paths(), + [format!("/{profile}/api/cluster/info")] + ); + } + let (server, client) = setup(profile); + let body = json!({"input":encode_value_envelope(&Value::Null, DEFAULT_CODEC).unwrap()}); + send(&client, "/workflows", false, &body).await.unwrap(); + assert_eq!( + server.request_body(&format!("/{profile}/api/workflows")), + body + ); + } +} + +#[tokio::test] +async fn completion_limits_do_not_enlarge_ordinary_or_failure_requests() { + for (path, worker, field) in [ + ("/workflows", false, "input"), + ("/worker/activity-tasks/test/complete", true, "result"), + ("/worker/query-tasks/test/complete", true, "result_envelope"), + ("/worker/activity-tasks/test/fail", true, "failure"), + ] { + let (server, client) = setup("unconfigured"); + let envelope = encode_value_envelope(&Value::Null, DEFAULT_CODEC).unwrap(); + let value = if field == "failure" { + json!({"details":envelope}) + } else { + envelope + }; + let body = json!({field:value, "metadata":"x".repeat(3000)}); + assert!(send(&client, path, worker, &body).await.is_err(), "{path}"); + assert_eq!(server.captured_paths(), ["/unconfigured/api/cluster/info"]); + assert_eq!( + client + .known_failure_request_limit(RequestProtocol::DefaultWorker) + .unwrap(), + 2048 + ); + } +} + +#[tokio::test] +async fn completion_limits_use_explicit_smaller_caps_and_plan_uploads_before_effects() { + for path in [WORKFLOW, CONTINUATION] { + let (server, client) = setup("available"); + send(&client, path, true, &batch(3)).await.unwrap(); + assert_eq!( + server.request_count("/available/api/external-payloads/v1"), + 0 + ); + send(&client, path, true, &batch(8)).await.unwrap(); + assert!(server.request_count("/available/api/external-payloads/v1") > 0); + assert!( + server.request_bodies(&format!("/available/api{path}"))[1] + .to_string() + .len() + <= 8192 + ); + + let (server, client) = setup("small"); + send(&client, path, true, &batch(1)).await.unwrap(); + assert_eq!(server.request_count("/small/api/external-payloads/v1"), 1); + assert!( + server + .request_body(&format!("/small/api{path}")) + .to_string() + .len() + <= 1024 + ); + let body = json!({"commands":batch(3)["commands"], "metadata":"x".repeat(1500)}); + let before = server.request_count("/small/api/external-payloads/v1"); + assert!(send(&client, path, true, &body).await.is_err()); + assert_eq!( + server.request_count("/small/api/external-payloads/v1"), + before + ); + } +} + +#[tokio::test] +async fn completion_limits_preserve_configured_storage_refusals() { + let (server, client) = setup("transient"); + let error = send(&client, WORKFLOW, true, &batch(8)).await.unwrap_err(); + assert!( + error.to_string().contains("external_payload_unavailable"), + "{error}" + ); + assert_eq!(server.captured_paths(), ["/transient/api/cluster/info"]); +} + +#[tokio::test] +async fn completion_limits_keep_authoritative_413_and_settle_without_retrying_the_batch() { + let (server, client) = setup("refused"); + let body = batch(8); + assert!(body.to_string().len() > 8192); + let error = send(&client, WORKFLOW, true, &body).await.unwrap_err(); + let Error::WorkflowTaskCompletionTooLarge(refusal) = error else { + panic!("{error:?}"); + }; + assert_eq!(refusal.status, 413); + assert_eq!(refusal.limit, Some(8192)); + assert_eq!(refusal.size, body.to_string().len() as u64); + assert_eq!(server.request_count("/refused/api/external-payloads/v1"), 0); + + let (server, client) = setup("refused"); + let worker = Worker::new(client, "fixture-queue"); + let commands = body["commands"].as_array().unwrap().clone(); + let outcome = worker + .settle_workflow_task_decision( + "test", + "worker", + 2, + Some("run"), + Ok(WorkflowTaskDecision::without_message_streams(commands)), + false, + None, + ) + .await + .unwrap(); + assert!(matches!(outcome, ManagedPollOutcome::Handled)); + let sent = server.request_bodies("/refused/api/worker/workflow-tasks/test/complete"); + assert_eq!(sent.len(), 2); + assert_eq!(sent[0]["commands"], body["commands"]); + assert_eq!(sent[1]["commands"].as_array().unwrap().len(), 1); + assert_eq!(sent[1]["commands"][0]["type"], "fail_workflow"); + assert_eq!(sent[1]["commands"][0]["non_retryable"], true); + assert_eq!(sent[1]["lease_owner"], "worker"); + assert_eq!(sent[1]["workflow_task_attempt"], 2); + assert_eq!( + server.request_count("/refused/api/worker/workflow-tasks/test/fail"), + 0 + ); +} From 5338f45bd773519f71c1dfcab85d4f13082fa6a4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:41:51 +0000 Subject: [PATCH 2/4] Bind shared completion negotiation vectors and acceptance scope --- Cargo.toml | 2 +- scripts/acceptance/installed-native/README.md | 22 ++ src/runtime_uploads.rs | 9 +- src/tests/completion_limits.rs | 204 +++++++++++++----- tests/completion_limits_server.rs | 167 ++++++++++++++ ...low-task-completion-limits-v1-vectors.json | 145 +++++++++++++ 6 files changed, 484 insertions(+), 65 deletions(-) create mode 100644 tests/completion_limits_server.rs create mode 100644 tests/fixtures/workflow-task-completion-limits-v1-vectors.json diff --git a/Cargo.toml b/Cargo.toml index fa0c7cef..bb5107ca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ readme = "README.md" rust-version = "1.86" keywords = ["durable", "workflow", "temporal", "worker", "sdk"] categories = ["api-bindings", "asynchronous"] -include = ["/src/**", "/schema/**", "/examples/**", "/Cargo.toml", "/README.md", "/CHANGELOG.md", "/LICENSE", "/tests/fixtures/replay-regressions/published-php-complete-explicit-backoff.json", "/tests/fixtures/sdk-metrics-contract.json", "/tests/fixtures/auth-configuration/static-credentials-v1.json"] +include = ["/src/**", "/schema/**", "/examples/**", "/Cargo.toml", "/README.md", "/CHANGELOG.md", "/LICENSE", "/tests/fixtures/replay-regressions/published-php-complete-explicit-backoff.json", "/tests/fixtures/sdk-metrics-contract.json", "/tests/fixtures/workflow-task-completion-limits-v1-vectors.json", "/tests/fixtures/auth-configuration/static-credentials-v1.json"] [package.metadata.durable-workflow] product-train = "4.15.0" diff --git a/scripts/acceptance/installed-native/README.md b/scripts/acceptance/installed-native/README.md index d0491f98..dc01ed5a 100644 --- a/scripts/acceptance/installed-native/README.md +++ b/scripts/acceptance/installed-native/README.md @@ -44,6 +44,28 @@ It does the following: `check_history.py` verifies each cell's durable history through the runtime's public HTTP API, independently of the SDK under test. +### Completion-limit feature acceptance + +`tests/completion_limits_server.rs` exercises the planner fix separately from +the lifecycle cells. Use an isolated published Server with the default limits, +no payload store and explicit `limits.max_workflow_task_completion_bytes`. +The test refuses other producers. Set the same fixture environment variables +as the metrics qualifier, including `DURABLE_WORKFLOW_LOCAL_ISOLATED=1` and a +loopback `DURABLE_WORKFLOW_SERVER_URL`, then run: + +```sh +cargo test --locked --all-features --test completion_limits_server -- --ignored --nocapture +``` + +For published acceptance, copy this test and its +`fixtures/workflow-task-completion-limits-v1-vectors.json` into a fresh consumer +whose dependency pins the exact registry crate, with no source patch. It checks +300 × 10 KB inline arguments, an over-limit batch's structural failure with no +partially scheduled activities, and another workflow on the same worker. +Record the package checksum, Server digest, runner commit and cleanup alongside +the result. Compiling the ignored test or using source dependencies is not +published-artifact acceptance. + ## Run Fetch tags first (`git fetch --tags origin`), then run: diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs index 4000a528..67d53920 100644 --- a/src/runtime_uploads.rs +++ b/src/runtime_uploads.rs @@ -172,7 +172,7 @@ impl Client { return Ok(()); } let policy = self.runtime_upload_policy(protocol).await?; - let completion_bytes = if is_command_completion(path, protocol) { + let completion_bytes = if is_workflow_completion(path, protocol) { policy .completion_request_bytes .map_err(|()| unsupported("invalid workflow completion request limit"))? @@ -541,7 +541,7 @@ fn replace(body: &mut Value, path: &str, reference: Value) -> Result<()> { Ok(()) } -fn is_command_completion(path: &str, protocol: RequestProtocol) -> bool { +fn is_workflow_completion(path: &str, protocol: RequestProtocol) -> bool { if !matches!( protocol, RequestProtocol::DefaultWorker | RequestProtocol::Worker(_) @@ -552,7 +552,7 @@ fn is_command_completion(path: &str, protocol: RequestProtocol) -> bool { let parts: Vec<_> = path.strip_prefix('/').unwrap_or(path).split('/').collect(); matches!(parts.as_slice(), ["worker", "workflow-tasks", task, "complete"] - | ["worker", "activity-tasks", task, "continuation", "complete"] if !task.is_empty()) + if !task.is_empty()) } // Only protocol-owned slots are eligible; arbitrary application maps are not envelopes. @@ -565,8 +565,7 @@ fn payload_paths(body: &Value, path: &str, protocol: RequestProtocol) -> Vec { + ["worker", "workflow-tasks", _, "complete"] => { for (index, command) in body["commands"] .as_array() .into_iter() diff --git a/src/tests/completion_limits.rs b/src/tests/completion_limits.rs index 9fc6ac7d..fcdef7aa 100644 --- a/src/tests/completion_limits.rs +++ b/src/tests/completion_limits.rs @@ -4,10 +4,30 @@ use super::*; const WORKFLOW: &str = "/worker/workflow-tasks/test/complete"; const CONTINUATION: &str = "/worker/activity-tasks/test/continuation/complete"; +fn vectors() -> Value { + // Server e69df7c37d675681ca301b9eedc2398bc1a3ac18, #1136/#1203. + serde_json::from_str(include_str!( + "../../tests/fixtures/workflow-task-completion-limits-v1-vectors.json" + )) + .unwrap() +} + fn responses(path: &str, body: &str, _number: usize) -> Option<(&'static str, String)> { let profile = path.split('/').nth(1).unwrap_or_default(); if path.ends_with("/api/cluster/info") { let mut info = policy(); + if let Some(id) = profile.strip_prefix("vector-") { + let vectors = vectors(); + let case = vectors["negotiation_cases"] + .as_array() + .unwrap() + .iter() + .find(|case| case["id"] == id) + .unwrap(); + info["limits"] = case["limits"].clone(); + info["namespace"]["external_payload_storage"] = json!({"status":"unconfigured"}); + return Some(("200 OK", info.to_string())); + } let limit = match profile { "null" => Value::Null, "zero" => json!(0), @@ -39,6 +59,16 @@ fn responses(path: &str, body: &str, _number: usize) -> Option<(&'static str, St } return Some(("200 OK", info.to_string())); } + if profile == "vector-positive-smaller" { + return Some(( + "413 Payload Too Large", + json!({ + "reason":"payload_too_large", "limit_kind":"workflow_task_completion_bytes", + "limit":1024 + }) + .to_string(), + )); + } if path.ends_with("/api/external-payloads/v1") { return Some(( "201 Created", @@ -65,6 +95,65 @@ fn responses(path: &str, body: &str, _number: usize) -> Option<(&'static str, St )) } +#[tokio::test] +async fn completion_limits_consume_shared_negotiation_vectors() { + let vectors = vectors(); + for case in vectors["negotiation_cases"].as_array().unwrap() { + let profile = format!("vector-{}", case["id"].as_str().unwrap()); + let (server, client) = setup(&profile); + let body = batch(3); + let outcome = send(&client, WORKFLOW, true, &body).await; + match case["expected"].as_str().unwrap() { + "ordinary_fallback" => { + outcome.unwrap(); + } + "completion_cap" + if body.to_string().len() as u64 <= case["completion_cap"].as_u64().unwrap() => + { + outcome.unwrap(); + } + "completion_cap" => { + let Error::WorkflowTaskCompletionTooLarge(refusal) = outcome.unwrap_err() else { + panic!("{profile}"); + }; + assert_eq!(refusal.limit, case["completion_cap"].as_u64()); + assert_eq!(refusal.size, body.to_string().len() as u64); + } + "refuse_completion_planning_before_upload_or_submission" => { + assert!(outcome + .unwrap_err() + .to_string() + .contains("invalid workflow completion request limit")); + assert_eq!( + server.captured_paths(), + [format!("/{profile}/api/cluster/info")] + ); + } + expected => panic!("unknown vector outcome {expected}"), + } + assert_eq!( + server.request_count(&format!("/{profile}/api/external-payloads/v1")), + 0 + ); + } +} + +#[tokio::test] +async fn completion_limits_leave_unsupported_raw_continuation_offload_unchanged() { + let (server, client) = setup("available"); + let body = json!({"result":batch(1)["commands"][0]["arguments"], + "continuation":{"commands":batch(8)["commands"]}}); + send(&client, CONTINUATION, true, &body).await.unwrap(); + assert_eq!( + server.captured_paths(), + [format!("/available/api{CONTINUATION}")] + ); + assert_eq!( + server.request_body(&format!("/available/api{CONTINUATION}")), + body + ); +} + fn setup(profile: &str) -> (MockWorkerServer, Client) { let server = MockWorkerServer::start_with_behavior(MockWorkerBehavior { request_override: Some(responses), @@ -102,18 +191,17 @@ async fn completion_limits_preserve_inline_atomic_batches_without_storage() { let body = batch(3); assert!(body.to_string().len() > 2048 && body.to_string().len() < 8192); for profile in ["unconfigured", "disabled", "unsupported"] { - for path in [WORKFLOW, CONTINUATION] { - let (server, client) = setup(profile); - send(&client, path, true, &body).await.unwrap(); - assert_eq!( - server.request_bodies(&format!("/{profile}/api{path}")), - [body.clone()] - ); - assert_eq!( - server.request_count(&format!("/{profile}/api/external-payloads/v1")), - 0 - ); - } + let path = WORKFLOW; + let (server, client) = setup(profile); + send(&client, path, true, &body).await.unwrap(); + assert_eq!( + server.request_bodies(&format!("/{profile}/api{path}")), + [body.clone()] + ); + assert_eq!( + server.request_count(&format!("/{profile}/api/external-payloads/v1")), + 0 + ); } } @@ -143,20 +231,19 @@ async fn completion_limits_validate_explicit_values_only_on_completion_routes() for profile in [ "null", "zero", "negative", "fraction", "string", "boolean", "list", "object", ] { - for path in [WORKFLOW, CONTINUATION] { - let (server, client) = setup(profile); - let error = send(&client, path, true, &batch(3)).await.unwrap_err(); - assert!( - error.to_string().contains( - "external_payload_unsupported: invalid workflow completion request limit" - ), - "{profile} {path}: {error}" - ); - assert_eq!( - server.captured_paths(), - [format!("/{profile}/api/cluster/info")] - ); - } + let path = WORKFLOW; + let (server, client) = setup(profile); + let error = send(&client, path, true, &batch(3)).await.unwrap_err(); + assert!( + error.to_string().contains( + "external_payload_unsupported: invalid workflow completion request limit" + ), + "{profile} {path}: {error}" + ); + assert_eq!( + server.captured_paths(), + [format!("/{profile}/api/cluster/info")] + ); let (server, client) = setup(profile); let body = json!({"input":encode_value_envelope(&Value::Null, DEFAULT_CODEC).unwrap()}); send(&client, "/workflows", false, &body).await.unwrap(); @@ -196,40 +283,39 @@ async fn completion_limits_do_not_enlarge_ordinary_or_failure_requests() { #[tokio::test] async fn completion_limits_use_explicit_smaller_caps_and_plan_uploads_before_effects() { - for path in [WORKFLOW, CONTINUATION] { - let (server, client) = setup("available"); - send(&client, path, true, &batch(3)).await.unwrap(); - assert_eq!( - server.request_count("/available/api/external-payloads/v1"), - 0 - ); - send(&client, path, true, &batch(8)).await.unwrap(); - assert!(server.request_count("/available/api/external-payloads/v1") > 0); - assert!( - server.request_bodies(&format!("/available/api{path}"))[1] - .to_string() - .len() - <= 8192 - ); + let path = WORKFLOW; + let (server, client) = setup("available"); + send(&client, path, true, &batch(3)).await.unwrap(); + assert_eq!( + server.request_count("/available/api/external-payloads/v1"), + 0 + ); + send(&client, path, true, &batch(8)).await.unwrap(); + assert!(server.request_count("/available/api/external-payloads/v1") > 0); + assert!( + server.request_bodies(&format!("/available/api{path}"))[1] + .to_string() + .len() + <= 8192 + ); - let (server, client) = setup("small"); - send(&client, path, true, &batch(1)).await.unwrap(); - assert_eq!(server.request_count("/small/api/external-payloads/v1"), 1); - assert!( - server - .request_body(&format!("/small/api{path}")) - .to_string() - .len() - <= 1024 - ); - let body = json!({"commands":batch(3)["commands"], "metadata":"x".repeat(1500)}); - let before = server.request_count("/small/api/external-payloads/v1"); - assert!(send(&client, path, true, &body).await.is_err()); - assert_eq!( - server.request_count("/small/api/external-payloads/v1"), - before - ); - } + let (server, client) = setup("small"); + send(&client, path, true, &batch(1)).await.unwrap(); + assert_eq!(server.request_count("/small/api/external-payloads/v1"), 1); + assert!( + server + .request_body(&format!("/small/api{path}")) + .to_string() + .len() + <= 1024 + ); + let body = json!({"commands":batch(3)["commands"], "metadata":"x".repeat(1500)}); + let before = server.request_count("/small/api/external-payloads/v1"); + assert!(send(&client, path, true, &body).await.is_err()); + assert_eq!( + server.request_count("/small/api/external-payloads/v1"), + before + ); } #[tokio::test] diff --git a/tests/completion_limits_server.rs b/tests/completion_limits_server.rs new file mode 100644 index 00000000..7eef83e6 --- /dev/null +++ b/tests/completion_limits_server.rs @@ -0,0 +1,167 @@ +//! Copy this test into a fresh registry-only consumer for published acceptance. +use durable_workflow::{ + json, Client, Error, ParallelOperation, ParallelResult, Value, Worker, WorkflowResultOptions, +}; +use std::time::Duration; + +fn events<'a>(history: &'a Value, kind: &'a str) -> impl Iterator { + history["history_events"] + .as_array() + .expect("exported history") + .iter() + .filter(move |event| event["type"] == kind) +} + +#[tokio::test] +#[ignore = "requires an isolated published Server with negotiated completion limits"] +async fn published_worker_completes_inline_fanout_and_fails_an_oversized_batch() { + assert_eq!( + std::env::var("DURABLE_WORKFLOW_LOCAL_ISOLATED").as_deref(), + Ok("1") + ); + let origin = std::env::var("DURABLE_WORKFLOW_SERVER_URL").unwrap(); + assert!(origin + .strip_prefix("http://127.0.0.1:") + .and_then(|port| port.parse::().ok()) + .is_some_and(|port| port > 0)); + let client = Client::builder(origin) + .token(Some(std::env::var("DURABLE_WORKFLOW_AUTH_TOKEN").unwrap())) + .build() + .unwrap(); + let info = client.cluster_info().await.unwrap(); + let vectors: Value = serde_json::from_str(include_str!( + "fixtures/workflow-task-completion-limits-v1-vectors.json" + )) + .unwrap(); + assert_eq!( + info["limits"]["max_payload_bytes"], + vectors["default_limits"]["max_payload_bytes"] + ); + assert_eq!( + info["limits"]["max_workflow_task_completion_bytes"], + vectors["default_limits"]["max_workflow_task_completion_bytes"], + "qualification requires the explicit published producer, never a version/multiplier guess" + ); + assert!( + matches!( + info["namespace"]["external_payload_storage"]["status"].as_str(), + Some("unconfigured" | "disabled" | "unsupported") + ), + "fixture must have no payload store" + ); + + let queue = format!("completion-limits-{}", durable_workflow::Uuid::new_v4()); + let mut worker = Worker::new(client.clone(), &queue).max_concurrent_activity_tasks(16); + worker.register_activity("limits.bytes", |_, args| async move { + Ok(json!(args[0].as_str().unwrap().len())) + }); + worker.register_workflow("limits.fanout", |ctx, args| async move { + let count = args[0].as_u64().unwrap() as usize; + let size = args[1].as_u64().unwrap() as usize; + let payload = "x".repeat(size); + let operations = (0..count) + .map(|_| ParallelOperation::activity("limits.bytes", json!([payload]))) + .collect(); + let results = ctx.join(operations).await?; + assert_eq!(results.len(), count); + assert!(results.iter().all( + |value| matches!(value, ParallelResult::Activity(value) if *value == json!(size)) + )); + Ok(json!({"count":count,"bytes_each":size})) + }); + worker.register().await.unwrap(); + let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); + let mut running = tokio::spawn(async move { + worker + .run_until(async { + let _ = stopped.await; + }) + .await + }); + let options = WorkflowResultOptions { + timeout: Duration::from_secs(90), + ..Default::default() + }; + let normal = client + .start_workflow( + "limits.fanout", + &queue, + &format!("{queue}-normal"), + json!([300, 10_000]), + ) + .await + .unwrap(); + let result = tokio::select! { + result = normal.result(options.clone()) => result.unwrap(), + exit = &mut running => panic!("worker stopped during normal fanout: {exit:?}"), + }; + assert_eq!(result, json!({"count":300,"bytes_each":10_000})); + let history = client + .export_workflow_history(&normal.workflow_id, normal.run_id.as_deref().unwrap()) + .await + .unwrap(); + assert_eq!(events(&history, "ActivityCompleted").count(), 300); + assert_eq!(events(&history, "WorkflowCompleted").count(), 1); + + // 60 individually valid arguments exceed the default aggregate allowance + // after Avro/base64 and JSON encoding, without exceeding pending-work caps. + let oversized = client + .start_workflow( + "limits.fanout", + &queue, + &format!("{queue}-oversized"), + json!([60, 450_000]), + ) + .await + .unwrap(); + let error = tokio::select! { + result = oversized.result(options.clone()) => result.unwrap_err(), + exit = &mut running => panic!("worker stopped instead of recording structural failure: {exit:?}"), + }; + let Error::WorkflowFailed(outcome) = error else { + panic!("{error:?}"); + }; + assert_eq!( + outcome.failure_category.as_deref(), + Some("structural_limit") + ); + assert_eq!(outcome.non_retryable, Some(true)); + assert!(outcome + .message + .as_deref() + .unwrap() + .starts_with("Structural limit exceeded: workflow_task_completion_bytes:")); + let history = client + .export_workflow_history(&oversized.workflow_id, oversized.run_id.as_deref().unwrap()) + .await + .unwrap(); + assert_eq!(events(&history, "WorkflowFailed").count(), 1); + assert_eq!( + events(&history, "ActivityScheduled").count(), + 0, + "the rejected batch must not partially commit" + ); + assert_eq!(events(&history, "ActivityCompleted").count(), 0); + + let followup = client + .start_workflow( + "limits.fanout", + &queue, + &format!("{queue}-followup"), + json!([1, 10]), + ) + .await + .unwrap(); + let result = tokio::select! { + result = followup.result(options) => result.unwrap(), + exit = &mut running => panic!("worker did not remain usable: {exit:?}"), + }; + assert_eq!(result, json!({"count":1,"bytes_each":10})); + stop.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(15), running) + .await + .unwrap() + .unwrap() + .unwrap(); + println!("PASS: 300x10KB inline fanout; typed aggregate structural failure with no partial activities; same-worker followup"); +} diff --git a/tests/fixtures/workflow-task-completion-limits-v1-vectors.json b/tests/fixtures/workflow-task-completion-limits-v1-vectors.json new file mode 100644 index 00000000..5f766ab6 --- /dev/null +++ b/tests/fixtures/workflow-task-completion-limits-v1-vectors.json @@ -0,0 +1,145 @@ +{ + "schema": "durable-workflow.workflow-task-completion-limits.v1", + "issue": "https://github.com/durable-workflow/server/issues/1136", + "measurement": "complete_encoded_json_http_request_body_bytes", + "default_limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": 33554432 + }, + "completion_routes": [ + "/api/worker/workflow-tasks/fixture-task/complete", + "/api/worker/activity-tasks/fixture-task/continuation/complete" + ], + "ordinary_route": "/api/workflows", + "wire_cases": [ + { + "id": "ordinary-exact", + "route_kind": "ordinary", + "wire_bytes": 2097152, + "size_admitted": true + }, + { + "id": "ordinary-over", + "route_kind": "ordinary", + "wire_bytes": 2097153, + "size_admitted": false + }, + { + "id": "completion-over-ordinary", + "route_kind": "completion", + "wire_bytes": 2097153, + "size_admitted": true + }, + { + "id": "completion-exact", + "route_kind": "completion", + "wire_bytes": 33554432, + "size_admitted": true + }, + { + "id": "completion-over", + "route_kind": "completion", + "wire_bytes": 33554433, + "size_admitted": false + } + ], + "oversized_response": { + "status": 413, + "reason": "payload_too_large", + "limit_kind": "workflow_task_completion_bytes", + "limit": 33554432 + }, + "scope": "Size admission alone. Authentication, lease fences, per-value Avro limits and command validation still apply; no durable work is admitted by a size refusal.", + "negotiation_cases": [ + { + "id": "absent", + "limits": { + "max_payload_bytes": 2097152 + }, + "expected": "ordinary_fallback" + }, + { + "id": "positive", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": 33554432 + }, + "expected": "completion_cap", + "completion_cap": 33554432 + }, + { + "id": "positive-smaller", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": 1024 + }, + "expected": "completion_cap", + "completion_cap": 1024 + }, + { + "id": "null", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": null + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "zero", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": 0 + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "negative", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": -1 + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "float", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": 32.5 + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "string", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": "33554432" + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "bool", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": true + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "object", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": {} + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + }, + { + "id": "array", + "limits": { + "max_payload_bytes": 2097152, + "max_workflow_task_completion_bytes": [] + }, + "expected": "refuse_completion_planning_before_upload_or_submission" + } + ], + "negotiation_scope": "Only workflow-task completion and activity-continuation completion consume this field. Ordinary routes ignore it. Explicit invalid values refuse planning without upload/submission; absence preserves the existing conservative fallback." +} From 3283d0425fce11810c130f07d31d93b561b1e595 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:47:00 +0000 Subject: [PATCH 3/4] Validate completion limits before empty payload shortcut --- src/runtime_uploads.rs | 8 ++++++-- src/tests/completion_limits.rs | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs index 67d53920..ca7f95cf 100644 --- a/src/runtime_uploads.rs +++ b/src/runtime_uploads.rs @@ -168,17 +168,21 @@ impl Client { payloads.push((path, blob.len())); } } - if payloads.is_empty() { + let workflow_completion = is_workflow_completion(path, protocol); + if payloads.is_empty() && !workflow_completion { return Ok(()); } let policy = self.runtime_upload_policy(protocol).await?; - let completion_bytes = if is_workflow_completion(path, protocol) { + let completion_bytes = if workflow_completion { policy .completion_request_bytes .map_err(|()| unsupported("invalid workflow completion request limit"))? } else { None }; + if payloads.is_empty() { + return Ok(()); + } let request_bytes = completion_bytes.unwrap_or(policy.request_bytes); // Plan every replacement before any upload. Aggregate JSON size matters, // even when each individual value is below the namespace threshold. diff --git a/src/tests/completion_limits.rs b/src/tests/completion_limits.rs index fcdef7aa..cc85e701 100644 --- a/src/tests/completion_limits.rs +++ b/src/tests/completion_limits.rs @@ -138,6 +138,39 @@ async fn completion_limits_consume_shared_negotiation_vectors() { } } +#[tokio::test] +async fn completion_limits_refuse_invalid_vectors_even_without_inline_payloads() { + let vectors = vectors(); + for case in vectors["negotiation_cases"] + .as_array() + .unwrap() + .iter() + .filter(|case| case["expected"] == "refuse_completion_planning_before_upload_or_submission") + { + let profile = format!("vector-{}", case["id"].as_str().unwrap()); + for body in [ + json!({"commands":[]}), + json!({"commands":[{"type":"start_timer", "timer_id":"timer", "delay_seconds":1}]}), + json!({"commands":[{"type":"schedule_activity", "arguments":{ + "codec":"avro", "external_payload":reference("fixture") + }}]}), + ] { + let (server, client) = setup(&profile); + let error = send(&client, WORKFLOW, true, &body).await.unwrap_err(); + assert!( + error + .to_string() + .contains("invalid workflow completion request limit"), + "{profile}: {error}" + ); + assert_eq!( + server.captured_paths(), + [format!("/{profile}/api/cluster/info")] + ); + } + } +} + #[tokio::test] async fn completion_limits_leave_unsupported_raw_continuation_offload_unchanged() { let (server, client) = setup("available"); From 831d76c59f4c2920c39e6cadbe59e1569557e062 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:48:19 +0000 Subject: [PATCH 4/4] Expect discovery for reference-only completion validation --- src/tests/runtime_uploads.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/tests/runtime_uploads.rs b/src/tests/runtime_uploads.rs index ec6f9b33..3343e100 100644 --- a/src/tests/runtime_uploads.rs +++ b/src/tests/runtime_uploads.rs @@ -856,7 +856,7 @@ async fn runtime_upload_low_level_references_are_strict_and_never_reuploaded() { .await .unwrap(); assert_eq!(server.request_count(UPLOAD), 0); - assert_eq!(server.request_count(DISCOVERY), 0); + assert_eq!(server.request_count(DISCOVERY), 1); for (path, bad) in [ ("/codec", json!("json")), ("/external_payload/reference_id", json!("../../other")), @@ -875,4 +875,6 @@ async fn runtime_upload_low_level_references_are_strict_and_never_reuploaded() { .is_err()); } assert_eq!(server.request_count(&format!("/api{COMPLETE}")), 1); + assert_eq!(server.request_count(DISCOVERY), 1); + assert_eq!(server.request_count(UPLOAD), 0); }