Deliver typed oversized completion refusals and advertise the wire cap

.github/workflows/server-parity.yml

@@ -27,6 +27,7 @@ on:
       - tests/Unit/NativeDebugArtifactTest.mjs
       - config/dw-contract.php
       - resources/platform-protocol-specs/carried-poll-v1-vectors.json
+      - resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json
       - tests/Fixtures/ActivityMetadataProjection/**
       - tests/Fixtures/CodecRegression/**
       - tests/Fixtures/CommandIntent/**

docs/server-reference.md

@@ -645,6 +645,29 @@ workflow-task command payload.
 dedicated matching-role loop should skip a pass rather than duplicate work
 already covered by another matching-role process holding the repair throttle.
 
+The native completion transport correction for [#1136](https://github.com/durable-workflow/server/issues/1136)
+adds `limits.max_workflow_task_completion_bytes` to cluster discovery. It is the
+actual cap on the complete encoded JSON HTTP request body for workflow-task
+completion and activity continuation completion, including envelopes, metadata
+and commands. The default is 33,554,432 bytes. `limits.max_payload_bytes` remains
+the per-value and ordinary API request limit; a decoded argument sum does not
+measure completion wire size. SDKs use the explicit optional field for those
+two completion routes and preserve their conservative fallback when it is absent.
+A positive integer smaller than the ordinary cap is still authoritative for
+completion. Explicit malformed, null, zero, negative or non-integer values
+refuse completion planning before upload or submission; ordinary routes ignore
+this unused field.
+
+A completion over that cap returns HTTP 413 with `reason: payload_too_large`,
+`limit_kind: workflow_task_completion_bytes` and the actual byte `limit`, before
+any workflow state changes. The server discards at most 64 MiB of remaining
+upload data for at most two seconds before closing the HTTP/1 connection. This
+lets ordinary within-budget uploads finish sending and receive the typed refusal;
+clients that exceed either discard bound can still encounter a transport failure.
+The [shared size vectors](../resources/platform-protocol-specs/workflow-task-completion-limits-v1-vectors.json)
+cover exact and over-cap bodies on both completion routes. Published SDK worker
+qualification remains required before claiming this correction delivered.
+
 ### Namespaces
 - `GET /api/namespaces` — List namespaces
 - `POST /api/namespaces` — Create namespace

resources/platform-protocol-specs/cluster-info-envelope.schema.json

@@ -174,6 +174,7 @@
       "additionalProperties": true,
       "properties": {
         "max_payload_bytes": {"type": "integer", "minimum": 1},
+        "max_workflow_task_completion_bytes": {"type": "integer", "minimum": 1, "description": "Optional actual encoded JSON HTTP request-body cap for workflow-task completion and activity continuation completion. Per-value max_payload_bytes still applies. Missing means the producer has not advertised a separate completion budget; consumers must preserve their conservative fallback and must not infer a multiplier from server version."},
         "max_memo_bytes": {"type": "integer", "minimum": 1},
         "max_search_attributes": {"type": "integer", "minimum": 0},
         "max_search_attribute_key_length": {"type": "integer", "minimum": 1},

resources/platform-protocol-specs/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."
+}

rust/src/runtime/payload_discovery.rs

@@ -104,6 +104,7 @@ pub(super) fn limits() -> Result<Value> {
     let attributes = super::search_attribute_upserts::Limits::from_env()?;
     Ok(json!({
         "max_payload_bytes": super::payload_limits::max_payload_bytes(),
+        "max_workflow_task_completion_bytes": super::payload_limits::max_completion_bytes(),
         "max_memo_bytes": super::schedules::max_memo_bytes(),
         "max_search_attributes": super::search_attributes::configured_limit()?,
         "max_search_attribute_key_length": attributes.key_length,

rust/src/runtime/payload_limits.rs

@@ -4,20 +4,26 @@
 //! refused with PHP's JSON error, in the worker or control-plane envelope.
 //! The external payload upload keeps its own transport rules.
 use axum::{
-    body::{Body, to_bytes},
+    body::{Body, BodyDataStream},
     extract::Request,
     http::{Method, StatusCode, header},
     middleware::Next,
     response::Response,
 };
 use serde_json::{Value, json};
+use futures_util::{StreamExt, stream};
+use std::time::Duration;
 
 /// How many payloads' worth of bytes one completion that carries workflow
 /// commands may send (Server #1136, .github#228). Each command's arguments
 /// still meet the payload limit inside the completion; this bounds only the
 /// request as a whole, so a fan-out whose commands are each within the limits
 /// is not refused as one oversized body.
 const COMPLETION_PAYLOADS: usize = 16;
+// Refusals must let ordinary clients finish their current upload before the
+// HTTP/1 connection closes. Bound both discarded traffic and slow senders.
+const REFUSAL_DRAIN_BYTES: usize = 64 * 1024 * 1024;
+const REFUSAL_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
 
 /// A request that completes a workflow task with commands.
 fn carries_commands(method: &Method, path: &str) -> bool {
@@ -37,20 +43,47 @@ pub(super) fn max_payload_bytes() -> usize {
         .unwrap_or(2 * 1024 * 1024)
 }
 
+pub(super) fn max_completion_bytes() -> usize {
+    max_payload_bytes().saturating_mul(COMPLETION_PAYLOADS)
+}
+
+fn oversized_response(response: Response, mut incoming: BodyDataStream) -> Response {
+    let (mut parts, body) = response.into_parts();
+    parts.headers.insert(header::CONNECTION, axum::http::HeaderValue::from_static("close"));
+    let response_stream = stream::once(async move {
+        let drain = async {
+            let mut consumed = 0_usize;
+            while let Some(Ok(chunk)) = incoming.next().await {
+                consumed = consumed.saturating_add(chunk.len());
+                if consumed >= REFUSAL_DRAIN_BYTES {
+                    break;
+                }
+            }
+        };
+        let _ = tokio::time::timeout(REFUSAL_DRAIN_TIMEOUT, drain).await;
+        // Keep the request body alive until the discard finishes. Dropping
+        // it with bytes in flight makes Hyper close its read side and can
+        // turn the typed 413 into a client write/read transport failure.
+        body.into_data_stream()
+    }).flatten();
+    Response::from_parts(parts, Body::from_stream(response_stream))
+}
+
 pub(super) async fn enforce(request: Request, next: Next) -> Response {
     let path = request.uri().path().to_owned();
     if request.method() == Method::POST && path == "/api/external-payloads/v1" {
         return next.run(request).await;
     }
     let completion = carries_commands(request.method(), &path);
-    let max = if completion { max_payload_bytes().saturating_mul(COMPLETION_PAYLOADS) } else { max_payload_bytes() };
+    let max = if completion { max_completion_bytes() } else { max_payload_bytes() };
     let worker = path == "/api/worker" || path.starts_with("/api/worker/");
     let declared = request
         .headers()
         .get(header::CONTENT_LENGTH)
         .and_then(|value| value.to_str().ok())
         .and_then(|value| value.trim().parse::<u64>().ok());
     let (parts, body) = request.into_parts();
+    let mut incoming = body.into_data_stream();
     let refuse = |status: StatusCode, body: Value| {
         let parts = parts.clone();
         async move { super::http::admission_error(Request::from_parts(parts, Body::empty()), worker, status, body).await }
@@ -67,13 +100,18 @@ pub(super) async fn enforce(request: Request, next: Next) -> Response {
         body
     };
     if declared.is_some_and(|declared| declared > max as u64) {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+        return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
     }
-    let Ok(bytes) = to_bytes(body, max.saturating_add(1)).await else {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
-    };
-    if bytes.len() > max {
-        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+    let mut bytes = Vec::new();
+    while let Some(chunk) = incoming.next().await {
+        let Ok(chunk) = chunk else {
+            return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
+        };
+        if chunk.len() > max.saturating_sub(bytes.len()) {
+            drop(bytes);
+            return oversized_response(refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await, incoming);
+        }
+        bytes.extend_from_slice(&chunk);
     }
     let has_body = !bytes.is_empty() || declared.is_some_and(|declared| declared > 0);
     if matches!(parts.method, Method::POST | Method::PUT | Method::PATCH | Method::DELETE) && has_body {

rust/tests/execution/payload_discovery.rs

@@ -62,6 +62,7 @@ async fn cluster_info_discovers_runtime_payload_storage_like_php() {
         assert_eq!(storage["transport"], info["worker_protocol"]["server_capabilities"]["runtime_external_payload_transport"],
             "{name} namespace transport is the worker capability");
         assert_eq!(info["limits"]["max_payload_bytes"], 2_097_152);
+        assert_eq!(info["limits"]["max_workflow_task_completion_bytes"], 33_554_432);
         assert_eq!(info["limits"]["max_memo_bytes"], 262_144);
         assert_eq!(info["limits"]["max_search_attributes"], 100);
         assert_eq!(info["default_namespace"], "default");