From 7db1f87f6c2a1576df6af2e85d329e327dd1eae3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 17:01:58 +0000 Subject: [PATCH 1/2] Retry draining payload uploads with negotiated completion context --- src/runtime_uploads.rs | 124 +++++++++++++++++-- src/tests/runtime_uploads.rs | 223 ++++++++++++++++++++++++++++++++++- 2 files changed, 332 insertions(+), 15 deletions(-) diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs index d0e76eb..ca73a12 100644 --- a/src/runtime_uploads.rs +++ b/src/runtime_uploads.rs @@ -2,6 +2,8 @@ use super::*; use runtime_payloads::Reference; pub(super) type PolicyCache = [Option<(Instant, Policy)>; 2]; +const COMPLETION_SCHEMA: &str = "durable-workflow.v2.payload-completion-context.v1"; +const COMPLETION_HEADER: &str = "x-durable-workflow-payload-completion"; #[derive(Clone, Debug)] pub(super) struct Policy { @@ -10,6 +12,7 @@ pub(super) struct Policy { request_bytes: usize, timeout: Duration, available: bool, + completion_context: bool, } fn unsupported(message: &str) -> Error { @@ -38,6 +41,7 @@ impl Policy { request_bytes, timeout: Duration::from_secs(30), available: false, + completion_context: false, }); } if manifest["schema"] != "durable-workflow.v2.runtime-external-payload-transport.v1" @@ -64,6 +68,10 @@ impl Policy { request_bytes, timeout: Duration::from_secs(timeout as u64), available: storage["status"] == "available", + completion_context: manifest["upload"]["completion_context"]["schema"] + == COMPLETION_SCHEMA + && manifest["upload"]["completion_context"]["header"] + == "X-Durable-Workflow-Payload-Completion", }) } } @@ -137,6 +145,13 @@ impl Client { let reference = if let Some(reference) = uploaded.get(&identity) { reference.clone() } else { + let completion = if policy.completion_context + && matches!(protocol, RequestProtocol::Worker(_)) + { + completion_context(body, path, &upload.path) + } else { + None + }; let request = self .runtime_payload_request( reqwest::Method::POST, @@ -152,7 +167,7 @@ impl Client { .body(upload.blob) .build()?; let response = self - .runtime_payload_json(request, protocol, 64 * 1024) + .runtime_payload_json(request, protocol, 64 * 1024, completion) .await?; if response["schema"] != "durable-workflow.v2.runtime-external-payload-upload.v1" || response["transport_version"] != 1 @@ -192,7 +207,7 @@ impl Client { .runtime_payload_request(reqwest::Method::GET, "/cluster/info", protocol, true)? .build()?; let info = self - .runtime_payload_json(request, protocol, 2 * 1024 * 1024) + .runtime_payload_json(request, protocol, 2 * 1024 * 1024, None) .await?; let policy = Policy::from_info(&info)?; self.runtime_upload_policy @@ -234,17 +249,22 @@ impl Client { request: reqwest::Request, protocol: RequestProtocol, limit: usize, + completion: Option, ) -> Result { let mut retries = 0; + let mut bound_retry = false; loop { - let mut response = self - .http - .execute( - request - .try_clone() - .ok_or_else(|| unsupported("upload body cannot be retried"))?, - ) - .await?; + let mut attempt = request + .try_clone() + .ok_or_else(|| unsupported("upload body cannot be retried"))?; + if bound_retry { + if let Some(context) = &completion { + attempt + .headers_mut() + .insert(COMPLETION_HEADER, context.clone()); + } + } + let mut response = self.http.execute(attempt).await?; let status = response.status(); if response .content_length() @@ -269,10 +289,43 @@ impl Client { return Err(Error::Protocol(failure)); } let error = Error::Http { status, body }; + if !bound_retry + && completion.is_some() + && status == reqwest::StatusCode::SERVICE_UNAVAILABLE + && worker_storage_admission_body(&error).is_some_and(|body| { + body["reason"] == "storage_pressure" && body["storage_state"] == "draining" + }) + { + bound_retry = true; + continue; + } + // Upload bytes are content-addressed; a late refusal may safely + // retry them. Never relax admission for ordinary mutations. + let retry_error = worker_storage_admission_body(&error).and_then(|mut body| { + if request.method() == reqwest::Method::POST + && request.url().path().ends_with("/api/external-payloads/v1") + && body.get("request_admitted").is_none() + { + body["request_admitted"] = json!(false); + Some(Error::Http { + status, + body: body.to_string(), + }) + } else { + None + } + }); if self - .wait_for_storage_admission(&error, protocol, None, &mut retries) + .wait_for_storage_admission( + retry_error.as_ref().unwrap_or(&error), + protocol, + None, + &mut retries, + ) .await { + // Try ordinary admission again after capacity recovers. + bound_retry = false; continue; } return Err(error); @@ -287,6 +340,55 @@ impl Client { } } +fn completion_context( + body: &Value, + path: &str, + slot: &str, +) -> Option { + let parts: Vec<_> = path + .split('?') + .next()? + .trim_start_matches('/') + .split('/') + .collect(); + let ["worker", family, task_id, operation] = parts.as_slice() else { + return None; + }; + let (kind, field) = match *family { + "activity-tasks" => ("activity", "activity_attempt_id"), + "workflow-tasks" => ("workflow", "workflow_task_attempt"), + "query-tasks" => ("query", "query_task_attempt"), + _ => return None, + }; + if !matches!(*operation, "complete" | "fail") || task_id.is_empty() { + return None; + } + let owner = body["lease_owner"].as_str().filter(|s| !s.is_empty())?; + let attempt = &body[field]; + if kind == "activity" { + attempt.as_str().filter(|s| !s.is_empty())?; + } else { + attempt.as_u64().filter(|n| *n > 0)?; + } + // These pointers come only from payload_paths, never from application maps. + let slot: Vec = slot + .trim_start_matches('/') + .split('/') + .map(|part| { + part.parse::() + .map(Value::from) + .unwrap_or_else(|_| json!(part)) + }) + .collect(); + let context = json!({"schema": COMPLETION_SCHEMA, "kind": kind, "task_id": task_id, + "attempt": attempt, "lease_owner": owner, "operation": operation, "slot": slot}) + .to_string(); + if context.len() > 4096 { + return None; + } + reqwest::header::HeaderValue::from_str(&context).ok() +} + fn select(body: &mut Value, path: &str) -> Result { let value = body .pointer_mut(path) diff --git a/src/tests/runtime_uploads.rs b/src/tests/runtime_uploads.rs index 381af98..587bee4 100644 --- a/src/tests/runtime_uploads.rs +++ b/src/tests/runtime_uploads.rs @@ -31,6 +31,19 @@ fn responses(path: &str, blob: &str, number: usize) -> Option<(&'static str, Str if path.ends_with(DISCOVERY) { let mut value = policy(); let storage = &mut value["namespace"]["external_payload_storage"]; + if path.starts_with("/lease-") && !path.starts_with("/lease-legacy/") { + storage["transport"]["upload"]["completion_context"] = json!({ + "schema":"durable-workflow.v2.payload-completion-context.v1", + "header":"X-Durable-Workflow-Payload-Completion" + }); + if path.starts_with("/lease-unknown/") { + storage["transport"]["upload"]["completion_context"]["schema"] = json!("unknown"); + } + if path.starts_with("/lease-header/") { + storage["transport"]["upload"]["completion_context"]["header"] = + json!("Unexpected-Header"); + } + } match path.split('/').nth(1).unwrap_or("") { "unavailable" => storage["status"] = json!("unavailable"), "aggregate" => storage["threshold_bytes"] = json!(1048576), @@ -57,10 +70,27 @@ fn responses(path: &str, blob: &str, number: usize) -> Option<(&'static str, Str return Some(("200 OK", value.to_string())); } if path.ends_with(UPLOAD) { - if path.starts_with("/pressure/") && number <= 2 { + if path.starts_with("/lease-") && !path.starts_with("/lease-healthy/") { + if number == 1 || (number == 2 && path.starts_with("/lease-budget/")) { + let mut refusal = storage_refusal(None, false, false); + refusal["storage_state"] = json!(if path.starts_with("/lease-fenced/") { + "fenced" + } else { + "draining" + }); + return Some(("503 Service Unavailable", refusal.to_string())); + } + if path.starts_with("/lease-rejected/") { + return Some(( + "409 Conflict", + json!({"reason":"external_payload_completion_lease_rejected"}).to_string(), + )); + } + } + if (path.starts_with("/pressure/") || path.starts_with("/pressure-late/")) && number <= 2 { return Some(( "503 Service Unavailable", - storage_refusal(None, false, false).to_string(), + storage_refusal(None, false, path.starts_with("/pressure-late/")).to_string(), )); } if path.starts_with("/stopped/") { @@ -125,6 +155,191 @@ async fn send(client: &Client, path: &str, worker: bool, body: Value) -> Result< .await } +fn completion_header(headers: &str) -> Option { + headers.lines().find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("X-Durable-Workflow-Payload-Completion") + .then(|| serde_json::from_str(value.trim()).unwrap()) + }) +} + +#[tokio::test] +async fn runtime_upload_draining_retry_binds_each_completion_payload_to_its_lease() { + let mut cases = vec![ + ( + "/worker/activity-tasks/test/complete", + "activity", + json!("attempt-a"), + json!({"activity_attempt_id":"attempt-a", "result":payload()}), + json!(["result"]), + ), + ( + "/worker/activity-tasks/test/fail", + "activity", + json!("attempt-a"), + json!({"activity_attempt_id":"attempt-a", "failure":{"details":payload()}}), + json!(["failure", "details"]), + ), + ( + "/worker/query-tasks/test/complete", + "query", + json!(3), + json!({"query_task_attempt":3, "result_envelope":payload()}), + json!(["result_envelope"]), + ), + ( + COMPLETE, + "workflow", + json!(2), + json!({"workflow_task_attempt":2, + "commands":[{"type":"fail_workflow", "exception":{"details":payload()}}]}), + json!(["commands", 0, "exception", "details"]), + ), + ( + COMPLETE, + "workflow", + json!(2), + json!({"workflow_task_attempt":2, + "commands":[{"type":"record_side_effect", "workflow_stream":{"items":[ + {"payload":payload()["blob"], "payload_codec":"avro"}]}}]}), + json!(["commands", 0, "workflow_stream", "items", 0, "payload"]), + ), + ]; + for kind in [ + "complete_workflow", + "complete_update", + "record_side_effect", + "schedule_activity", + "start_child_workflow", + "continue_as_new", + "start_service_operation", + "upsert_memo", + ] { + let field = workflow_command_payload_field(kind).unwrap(); + cases.push(( + COMPLETE, + "workflow", + json!(2), + json!({"workflow_task_attempt":2, + "commands":[{"type":kind, field:payload()}]}), + json!(["commands", 0, field]), + )); + } + for (path, kind, attempt, mut body, slot) in cases { + let server = server(); + let client = Client::builder(format!("{}/lease-draining", server.base_url())) + .worker_token(Some("worker-only".into())) + .namespace("tenant-a") + .build() + .unwrap(); + body["lease_owner"] = json!("Worker-A"); + send(&client, path, true, body).await.unwrap(); + let requests = server.requests.lock().unwrap(); + let uploads: Vec<_> = requests + .iter() + .filter(|r| r.path.ends_with(UPLOAD)) + .collect(); + assert_eq!(uploads.len(), 2, "{path}: {slot}"); + assert_eq!(uploads[0].body, uploads[1].body); + assert!(completion_header(&uploads[0].headers).is_none()); + assert_eq!( + completion_header(&uploads[1].headers), + Some(json!({ + "schema":"durable-workflow.v2.payload-completion-context.v1", "kind":kind, + "task_id":"test", "attempt":attempt, "lease_owner":"Worker-A", + "operation":path.rsplit('/').next().unwrap(), "slot":slot + })) + ); + assert_eq!(uploads[1].namespace.as_deref(), Some("tenant-a")); + assert_eq!( + uploads[1].authorization.as_deref(), + Some("Bearer worker-only") + ); + assert_eq!( + uploads[1].worker_protocol.as_deref(), + Some(WORKER_PROTOCOL_VERSION) + ); + } +} + +#[tokio::test] +async fn runtime_upload_drain_capability_never_bypasses_unsupported_or_fenced_admission() { + for prefix in [ + "lease-legacy", + "lease-unknown", + "lease-header", + "lease-fenced", + "lease-rejected", + "lease-healthy", + "lease-client", + "lease-missing-attempt", + ] { + let server = server(); + let client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); + let worker = prefix != "lease-client"; + let path = if worker { COMPLETE } else { "/workflows" }; + let mut body = json!({"lease_owner":"worker", "workflow_task_attempt":1, + "commands":[{"type":"complete_workflow", "result":payload()}]}); + if !worker { + body = json!({"input":payload()}); + } + if prefix == "lease-missing-attempt" { + body["workflow_task_attempt"] = Value::Null; + } + let result = send(&client, path, worker, body).await; + if prefix == "lease-healthy" { + result.unwrap(); + } else { + let error = result.unwrap_err(); + assert!(matches!(error, Error::Http { .. }), "{prefix}: {error}"); + assert_eq!(server.request_count(&format!("/{prefix}/api{path}")), 0); + } + let requests = server.requests.lock().unwrap(); + let uploads: Vec<_> = requests + .iter() + .filter(|r| r.path.ends_with(UPLOAD)) + .collect(); + assert_eq!( + uploads.len(), + if prefix == "lease-rejected" { 2 } else { 1 }, + "{prefix}" + ); + assert!(completion_header(&uploads[0].headers).is_none()); + } +} + +#[tokio::test] +async fn runtime_upload_drain_budget_refusal_resumes_ordinary_upload_after_capacity_recovers() { + let server = server(); + let mut client = Client::new(format!("{}/lease-budget", server.base_url())).unwrap(); + client.worker_storage_admission = Some(WorkerStorageAdmission { + stop: Arc::new(AtomicBool::new(false)), + policy: WorkerRetryPolicy { + max_backoff: Duration::from_millis(1), + ..WorkerRetryPolicy::default() + }, + }); + send( + &client, + COMPLETE, + true, + json!({"lease_owner":"worker", "workflow_task_attempt":1, + "commands":[{"type":"complete_workflow", "result":payload()}]}), + ) + .await + .unwrap(); + let requests = server.requests.lock().unwrap(); + let uploads: Vec<_> = requests + .iter() + .filter(|r| r.path.ends_with(UPLOAD)) + .collect(); + assert_eq!(uploads.len(), 3); + assert!(uploads.iter().all(|r| r.body == uploads[0].body)); + assert!(completion_header(&uploads[0].headers).is_none()); + assert!(completion_header(&uploads[1].headers).is_some()); + assert!(completion_header(&uploads[2].headers).is_none()); +} + #[tokio::test] async fn runtime_upload_preserves_encoded_types_role_headers_and_base_path() { let server = server(); @@ -593,7 +808,7 @@ async fn runtime_upload_bounds_chunked_responses_and_does_not_follow_location() #[tokio::test] async fn runtime_upload_storage_admission_reuses_identical_bytes_and_shutdown_interrupts() { - for prefix in ["pressure", "stopped"] { + for prefix in ["pressure", "pressure-late", "stopped"] { let server = server(); let mut client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); let stop = Arc::new(AtomicBool::new(prefix == "stopped")); @@ -611,7 +826,7 @@ async fn runtime_upload_storage_admission_reuses_identical_bytes_and_shutdown_in json!({"commands":[{"type":"complete_workflow","result":payload()}]}), ) .await; - if prefix == "pressure" { + if prefix != "stopped" { outcome.unwrap(); assert_identical_requests(&server, &format!("/{prefix}{UPLOAD}"), 3); assert_eq!(server.request_count(&format!("/{prefix}/api{COMPLETE}")), 1); From 4c664ec89922c106c08834d5eb6de4dff72a098e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 17:37:08 +0000 Subject: [PATCH 2/2] Prepare Rust SDK 2.0.5 --- CHANGELOG.md | 9 +++++++++ Cargo.toml | 4 ++-- scripts/ci/test-publish-rust-sdk.py | 2 +- 3 files changed, 12 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35cf4f4..ab7653e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## 2.0.5 + +- On supporting Servers, retry a draining refusal for a completion payload + with its exact activity, workflow, or query lease and immutable payload slot. + Client uploads, unknown capabilities, and hard storage fences remain blocked. +- Preserve prepared completion bytes during late upload pressure without + re-executing handlers. Existing retries remain interruptible; Server upload + allowances, namespace quotas, and stale-lease errors remain authoritative. + ## 2.0.4 - Upload large encoded payloads through the authenticated namespace runtime diff --git a/Cargo.toml b/Cargo.toml index 21e9f31..8d98da2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "durable-workflow" -version = "2.0.4" +version = "2.0.5" edition = "2021" description = "Rust client and worker SDK for Durable Workflow Cloud and self-hosted Server" license = "MIT" @@ -14,7 +14,7 @@ categories = ["api-bindings", "asynchronous"] include = ["/src/**", "/schema/**", "/examples/**", "/Cargo.toml", "/README.md", "/CHANGELOG.md", "/LICENSE"] [package.metadata.durable-workflow] -product-train = "2.0.4" +product-train = "2.0.5" compatibility-authority = "protocol-manifests" supported-server-versions = "2.0.0" qualified-server-version = "2.0.0" diff --git a/scripts/ci/test-publish-rust-sdk.py b/scripts/ci/test-publish-rust-sdk.py index 907318b..a3b03d0 100644 --- a/scripts/ci/test-publish-rust-sdk.py +++ b/scripts/ci/test-publish-rust-sdk.py @@ -18,7 +18,7 @@ PUBLISH = ROOT / "scripts" / "ci" / "publish-rust-sdk.sh" RELEASE_WORKFLOW = ROOT / ".github" / "workflows" / "release.yml" RELEASE_TOOLING_INSTALLER = ROOT / "scripts" / "ci" / "install-release-tooling.sh" -PACKAGE_VERSION = "2.0.4" +PACKAGE_VERSION = "2.0.5" PRODUCT_TRAIN = PACKAGE_VERSION SERVER_VERSIONS = "2.0.0" QUALIFIED_SERVER_VERSION = "2.0.0"