Respect advertised completion limits for inline command batches

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.
+
 - Worker metrics emit the revision-2 schedule-to-start histogram in seconds
   for workflow and activity claims when the Server advertises
   `claimed_task_ready_age`. Missing or invalid producer values are omitted.

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"

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

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:

src/lib.rs

@@ -19758,6 +19758,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;

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<Option<usize>, ()>,
+    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"]
@@ -154,10 +168,22 @@ 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 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.
         payloads.sort_by_key(|(_, size)| std::cmp::Reverse(*size));
@@ -166,22 +192,34 @@ 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 {
                 uploads.push(select(body, path)?);
             }
         }
         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 +545,20 @@ fn replace(body: &mut Value, path: &str, reference: Value) -> Result<()> {
     Ok(())
 }
 
+fn is_workflow_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"]
+        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<String> {
     let path = path.split('?').next().unwrap_or(path);