From b9528f3c6683f35dd7e280f18c13452829206046 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 11:10:58 +0000 Subject: [PATCH 1/2] Upload Rust payloads through the authenticated namespace runtime --- src/lib.rs | 105 ++++++-- src/runtime_payloads.rs | 12 +- src/runtime_uploads.rs | 401 ++++++++++++++++++++++++++++ src/tests/runtime_uploads.rs | 498 +++++++++++++++++++++++++++++++++++ 4 files changed, 981 insertions(+), 35 deletions(-) create mode 100644 src/runtime_uploads.rs create mode 100644 src/tests/runtime_uploads.rs diff --git a/src/lib.rs b/src/lib.rs index 5c9e6fe..4d1add3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,7 @@ #![doc = include_str!("../README.md")] mod runtime_payloads; +mod runtime_uploads; use std::{ any::{type_name, Any, TypeId}, @@ -2272,7 +2273,9 @@ fn validate_workflow_task_commands(commands: &[Value]) -> Result<()> { let payload = command .get(payload_field) .ok_or_else(invalid_payload_envelope)?; - validate_outbound_payload_envelope(payload)?; + if runtime_payloads::Reference::parse(payload)?.is_none() { + validate_outbound_payload_envelope(payload)?; + } } Ok(()) } @@ -2622,6 +2625,7 @@ pub struct Client { namespace: String, max_external_payload_bytes: usize, worker_storage_admission: Option, + runtime_upload_policy: Arc>, } impl Client { @@ -4034,7 +4038,7 @@ impl Client { let auth_token = self.auth_token(protocol)?; let mut request = self .http - .request(method, format!("{}/api{}", self.base_url, path)) + .request(method.clone(), format!("{}/api{}", self.base_url, path)) .timeout(timeout) .header(reqwest::header::ACCEPT, "application/json") .header(reqwest::header::CONTENT_TYPE, "application/json") @@ -4057,14 +4061,18 @@ impl Client { } if let Some(body) = body { - request = request.json(body); + let mut body = serde_json::to_value(body)?; + if matches!( + method, + reqwest::Method::POST | reqwest::Method::PUT | reqwest::Method::PATCH + ) { + self.externalize_runtime_payloads(&mut body, path, protocol) + .await?; + } + request = request.json(&body); } let request = request.build()?; - let admission = self - .worker_storage_admission - .as_ref() - .filter(|_| matches!(protocol, RequestProtocol::Worker(_))); let poll_request_id = path.ends_with("/poll").then(|| { request .body() @@ -4092,28 +4100,16 @@ impl Client { return Err(Error::Protocol(protocol)); } let error = Error::Http { status, body }; - if let Some(admission) = admission { - if let Some(advertised_delay) = - worker_storage_admission_retry_after(&error, poll_request_id.as_deref()) - { - storage_retries = storage_retries.saturating_add(1); - let delay = worker_retry_delay(admission.policy, storage_retries) - .max(advertised_delay) - .min(admission.policy.max_backoff.max(Duration::from_millis(1))); - let deadline = tokio::time::Instant::now() + delay; - loop { - if admission.stop.load(Ordering::SeqCst) { - return Err(error); - } - let remaining = - deadline.saturating_duration_since(tokio::time::Instant::now()); - if remaining.is_zero() { - break; - } - tokio::time::sleep(remaining.min(Duration::from_millis(100))).await; - } - continue; - } + if self + .wait_for_storage_admission( + &error, + protocol, + poll_request_id.as_deref(), + &mut storage_retries, + ) + .await + { + continue; } return Err(error); } @@ -4129,6 +4125,41 @@ impl Client { } } + async fn wait_for_storage_admission( + &self, + error: &Error, + protocol: RequestProtocol, + poll_request_id: Option<&str>, + retries: &mut usize, + ) -> bool { + let Some(admission) = self + .worker_storage_admission + .as_ref() + .filter(|_| matches!(protocol, RequestProtocol::Worker(_))) + else { + return false; + }; + let Some(advertised_delay) = worker_storage_admission_retry_after(error, poll_request_id) + else { + return false; + }; + *retries = retries.saturating_add(1); + let delay = worker_retry_delay(admission.policy, *retries) + .max(advertised_delay) + .min(admission.policy.max_backoff.max(Duration::from_millis(1))); + let deadline = tokio::time::Instant::now() + delay; + loop { + if admission.stop.load(Ordering::SeqCst) { + return false; + } + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + return true; + } + tokio::time::sleep(remaining.min(Duration::from_millis(100))).await; + } + } + async fn poll_request_json( &self, path: &str, @@ -4637,6 +4668,7 @@ impl ClientBuilder { namespace: self.namespace, max_external_payload_bytes: self.max_external_payload_bytes, worker_storage_admission: None, + runtime_upload_policy: Arc::new(Mutex::new([None, None])), }) } } @@ -14727,6 +14759,7 @@ fn value_as_u64(value: &Value) -> Option { mod tests { use super::*; mod runtime_payloads; + mod runtime_uploads; use std::{ fs, io::{Read, Write}, @@ -24320,6 +24353,7 @@ mod tests { #[derive(Clone, Debug)] struct CapturedRequest { + headers: String, method: String, path: String, authorization: Option, @@ -24340,6 +24374,7 @@ mod tests { #[derive(Clone, Copy, Default)] struct MockWorkerBehavior { response_override: Option Option<(&'static str, String)>>, + request_override: Option Option<(&'static str, String)>>, storage_refusals: usize, storage_path: Option<&'static str>, storage_unavailable: bool, @@ -24766,6 +24801,10 @@ mod tests { let request_number = { let mut requests = requests.lock().expect("captured requests"); requests.push(CapturedRequest { + headers: request + .split_once("\r\n\r\n") + .map_or("", |(headers, _)| headers) + .to_owned(), method: method.to_string(), path: path.to_string(), authorization, @@ -24781,6 +24820,13 @@ mod tests { .count() }; + if let Some(response) = behavior + .request_override + .and_then(|handler| handler(path, body, request_number)) + { + write_mock_response(stream, response.0, &response.1); + return; + } if let Some(response) = behavior.response_override.and_then(|handler| handler(path)) { write_mock_response(stream, response.0, &response.1); return; @@ -25231,6 +25277,7 @@ mod tests { } let (status, body) = match path { + "/api/cluster/info" => ("200 OK", r#"{"limits":{"max_payload_bytes":2097152}}"#), "/api/health" => ("200 OK", r#"{"status":"ok"}"#), "/api/workflows" => ( "201 Created", diff --git a/src/runtime_payloads.rs b/src/runtime_payloads.rs index eb48b33..5b2ae2d 100644 --- a/src/runtime_payloads.rs +++ b/src/runtime_payloads.rs @@ -1,15 +1,15 @@ use super::*; -const SCHEMA: &str = "durable-workflow.v2.runtime-external-payload-reference.v1"; +pub(super) const SCHEMA: &str = "durable-workflow.v2.runtime-external-payload-reference.v1"; -#[derive(Deserialize, Eq, PartialEq, Hash)] +#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Hash)] #[serde(deny_unknown_fields)] -struct Reference { +pub(super) struct Reference { schema: String, codec: String, reference_id: String, - size_bytes: usize, - sha256: String, + pub(super) size_bytes: usize, + pub(super) sha256: String, } fn invalid_reference() -> Error { @@ -30,7 +30,7 @@ fn integrity_mismatch() -> Error { } impl Reference { - fn parse(envelope: &Value) -> Result> { + pub(super) fn parse(envelope: &Value) -> Result> { let Some(raw) = envelope.get("external_payload") else { return Ok(None); }; diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs new file mode 100644 index 0000000..dd53459 --- /dev/null +++ b/src/runtime_uploads.rs @@ -0,0 +1,401 @@ +use super::*; +use runtime_payloads::Reference; + +pub(super) type PolicyCache = [Option<(Instant, Policy)>; 2]; + +#[derive(Clone, Debug)] +pub(super) struct Policy { + threshold: usize, + max_bytes: usize, + request_bytes: usize, + timeout: Duration, + available: bool, +} + +fn unsupported(message: &str) -> Error { + Error::Codec(format!("external_payload_unsupported: {message}")) +} + +impl Policy { + fn from_info(info: &Value) -> Result { + let positive = |value: &Value| { + value + .as_u64() + .and_then(|n| usize::try_from(n).ok()) + .filter(|n| *n > 0) + }; + let request_bytes = if let Some(limit) = info.pointer("/limits/max_payload_bytes") { + positive(limit).ok_or_else(|| unsupported("invalid ordinary request limit"))? + } else { + 2 * 1024 * 1024 + }; + let storage = &info["namespace"]["external_payload_storage"]; + let manifest = &storage["transport"]; + if manifest.is_null() { + return Ok(Self { + threshold: request_bytes, + max_bytes: request_bytes, + request_bytes, + timeout: Duration::from_secs(30), + available: false, + }); + } + if manifest["schema"] != "durable-workflow.v2.runtime-external-payload-transport.v1" + || manifest["version"] != 1 + || manifest["reference_schema"] != runtime_payloads::SCHEMA + || manifest["mode"] != "authenticated_namespace_runtime" + || manifest["upload"]["method"] != "POST" + || manifest["upload"]["path"] != "/api/external-payloads/v1" + || manifest["fetch"]["method"] != "GET" + || manifest["fetch"]["path_template"] != "/api/external-payloads/v1/{referenceId}" + { + return Err(unsupported("invalid runtime transport manifest")); + } + let threshold = positive(&storage["threshold_bytes"]) + .ok_or_else(|| unsupported("invalid inline threshold"))?; + let max_bytes = positive(&manifest["limits"]["max_payload_bytes"]) + .filter(|max| *max >= threshold) + .ok_or_else(|| unsupported("invalid upload limit"))?; + let timeout = positive(&manifest["limits"]["request_timeout_seconds"]) + .ok_or_else(|| unsupported("invalid upload timeout"))?; + Ok(Self { + threshold, + max_bytes, + request_bytes, + timeout: Duration::from_secs(timeout as u64), + available: storage["status"] == "available", + }) + } +} + +struct Upload { + path: String, + blob: String, + sha256: String, +} + +impl Client { + pub(super) async fn externalize_runtime_payloads( + &self, + body: &mut Value, + path: &str, + protocol: RequestProtocol, + ) -> Result<()> { + let mut payloads = Vec::new(); + for path in payload_paths(body, path, protocol) { + let Some(value) = body.pointer(&path) else { + continue; + }; + Reference::parse(value)?; + let blob = value.as_str().or_else(|| { + (value.as_object().is_some_and(|map| map.len() == 2) + && value["codec"] == DEFAULT_CODEC) + .then(|| value["blob"].as_str()) + .flatten() + }); + if let Some(blob) = blob { + payloads.push((path, blob.len())); + } + } + if payloads.is_empty() { + return Ok(()); + } + let policy = self.runtime_upload_policy(protocol).await?; + // 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)); + if payloads.iter().any(|(_, size)| *size > policy.max_bytes) { + return Err(Error::Codec( + "external_payload_oversized: encoded payload exceeds runtime upload limit".into(), + )); + } + 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 { + 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 !uploads.is_empty() && !policy.available { + return Err(Error::Codec( + "external_payload_unavailable: namespace runtime storage is not available".into(), + )); + } + let mut uploaded = HashMap::<(String, usize), Reference>::new(); + for upload in uploads { + let identity = (upload.sha256.clone(), upload.blob.len()); + let reference = if let Some(reference) = uploaded.get(&identity) { + reference.clone() + } else { + let request = self + .runtime_payload_request( + reqwest::Method::POST, + "/external-payloads/v1", + protocol, + false, + )? + .timeout(policy.timeout) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .header("X-Durable-Workflow-Payload-Codec", DEFAULT_CODEC) + .header("X-Durable-Workflow-Payload-Size", upload.blob.len()) + .header("X-Durable-Workflow-Payload-SHA256", &upload.sha256) + .body(upload.blob) + .build()?; + let response = self + .runtime_payload_json(request, protocol, 64 * 1024) + .await?; + if response["schema"] != "durable-workflow.v2.runtime-external-payload-upload.v1" + || response["transport_version"] != 1 + { + return Err(unsupported("invalid upload response")); + } + let reference = Reference::parse( + &json!({"codec": DEFAULT_CODEC, "external_payload": response["reference"]}), + )? + .ok_or_else(|| unsupported("missing upload reference"))?; + if reference.size_bytes != identity.1 || reference.sha256 != identity.0 { + return Err(Error::Codec("external_payload_integrity_mismatch: upload response differs from submitted bytes".into())); + } + uploaded.insert(identity, reference.clone()); + reference + }; + replace(body, &upload.path, serde_json::to_value(reference)?)?; + } + Ok(()) + } + + async fn runtime_upload_policy(&self, protocol: RequestProtocol) -> Result { + let role = usize::from(matches!(protocol, RequestProtocol::Worker(_))); + let cached = self + .runtime_upload_policy + .lock() + .map_err(|_| unsupported("discovery cache poisoned"))?[role] + .clone(); + if let Some((checked, policy)) = cached { + if checked.elapsed() < Duration::from_secs(60) { + return Ok(policy); + } + } + // Discovery uses the control-plane header with this operation's credential + // role, so a worker never needs the application's client token. + let request = self + .runtime_payload_request(reqwest::Method::GET, "/cluster/info", protocol, true)? + .build()?; + let info = self + .runtime_payload_json(request, protocol, 512 * 1024) + .await?; + let policy = Policy::from_info(&info)?; + self.runtime_upload_policy + .lock() + .map_err(|_| unsupported("discovery cache poisoned"))?[role] = + Some((Instant::now(), policy.clone())); + Ok(policy) + } + + fn runtime_payload_request( + &self, + method: reqwest::Method, + path: &str, + protocol: RequestProtocol, + discovery: bool, + ) -> Result { + let mut request = self + .http + .request(method, format!("{}/api{path}", self.base_url)) + .header(reqwest::header::ACCEPT, "application/json") + .header("X-Namespace", &self.namespace); + request = match protocol { + RequestProtocol::Worker(version) if !discovery => { + request.header("X-Durable-Workflow-Protocol-Version", version) + } + _ => request.header( + "X-Durable-Workflow-Control-Plane-Version", + CONTROL_PLANE_VERSION, + ), + }; + if let Some(token) = self.auth_token(protocol)? { + request = request.bearer_auth(token); + } + Ok(request) + } + + async fn runtime_payload_json( + &self, + request: reqwest::Request, + protocol: RequestProtocol, + limit: usize, + ) -> Result { + let mut retries = 0; + loop { + let mut response = self + .http + .execute( + request + .try_clone() + .ok_or_else(|| unsupported("upload body cannot be retried"))?, + ) + .await?; + let status = response.status(); + if response + .content_length() + .is_some_and(|size| size > limit as u64) + { + return Err(unsupported( + "runtime transport response exceeds its byte limit", + )); + } + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await? { + if chunk.len() > limit.saturating_sub(bytes.len()) { + return Err(unsupported( + "runtime transport response exceeds its byte limit", + )); + } + bytes.extend_from_slice(&chunk); + } + if !status.is_success() { + let error = Error::Http { + status, + body: String::from_utf8_lossy(&bytes).into_owned(), + }; + if self + .wait_for_storage_admission(&error, protocol, None, &mut retries) + .await + { + continue; + } + return Err(error); + } + let value: Value = serde_json::from_slice(&bytes) + .map_err(|_| unsupported("runtime transport response is not JSON"))?; + if !value.is_object() { + return Err(unsupported("runtime transport response must be an object")); + } + return Ok(value); + } + } +} + +fn select(body: &mut Value, path: &str) -> Result { + let value = body + .pointer_mut(path) + .ok_or_else(|| unsupported("missing payload slot"))? + .take(); + let blob = match value { + Value::String(blob) => blob, + Value::Object(mut map) => match map.remove("blob") { + Some(Value::String(blob)) => blob, + _ => return Err(unsupported("missing encoded blob")), + }, + _ => return Err(unsupported("invalid payload slot")), + }; + let sha256 = format!("{:x}", Sha256::digest(blob.as_bytes())); + replace( + body, + path, + json!({"schema": runtime_payloads::SCHEMA, "codec": DEFAULT_CODEC, + "reference_id": "ep_00000000000000000000000000", "size_bytes": blob.len(), "sha256": sha256}), + )?; + Ok(Upload { + path: path.to_owned(), + blob, + sha256, + }) +} + +fn replace(body: &mut Value, path: &str, reference: Value) -> Result<()> { + let (parent, field) = path + .rsplit_once('/') + .ok_or_else(|| unsupported("invalid payload path"))?; + let object = body + .pointer_mut(parent) + .and_then(Value::as_object_mut) + .ok_or_else(|| unsupported("missing payload container"))?; + if field == "payload" { + object.remove("payload"); + object.insert("payload_reference".into(), reference); + object.insert("payload_codec".into(), json!(DEFAULT_CODEC)); + } else { + object.insert( + field.to_owned(), + json!({"codec": DEFAULT_CODEC, "external_payload": reference}), + ); + if field == "result_envelope" { + object.insert("result".into(), Value::Null); + } + } + Ok(()) +} + +// 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); + let parts: Vec<_> = path.trim_start_matches('/').split('/').collect(); + let mut paths = Vec::new(); + if matches!(protocol, RequestProtocol::Worker(_)) { + match parts.as_slice() { + ["worker", "workflow-tasks", _, "complete"] => { + for (index, command) in body["commands"] + .as_array() + .into_iter() + .flatten() + .enumerate() + { + let command_type = command["type"].as_str().unwrap_or_default(); + if let Some(field) = workflow_command_payload_field(command_type) { + paths.push(format!("/commands/{index}/{field}")); + } + if command_type == "record_local_activity" { + paths.extend([ + format!("/commands/{index}/arguments"), + format!("/commands/{index}/result"), + ]); + } + if command_type == "fail_workflow" { + paths.push(format!("/commands/{index}/exception/details")); + } + for item in 0..command["workflow_stream"]["items"] + .as_array() + .map_or(0, Vec::len) + { + paths.push(format!( + "/commands/{index}/workflow_stream/items/{item}/payload" + )); + } + } + } + ["worker", "activity-tasks", _, "complete"] => paths.push("/result".into()), + ["worker", "activity-tasks", _, "fail"] => paths.push("/failure/details".into()), + ["worker", "query-tasks", _, "complete"] => paths.push("/result_envelope".into()), + _ => {} + } + } else { + match parts.as_slice() { + ["workflows" | "activities"] + | ["workflows", _, "signal" | "query" | "update", _] + | ["workflows", _, "runs", _, "signal" | "query" | "update", _] + | ["workflows", _, "message-streams", _, "messages"] => paths.push("/input".into()), + ["schedules"] | ["schedules", _] => paths.push("/action/input".into()), + ["service-endpoints", _, "services", _, "operations", _, "execute"] => { + paths.push("/arguments".into()) + } + ["workflows", _, "runs", _, "streams", _, "items"] => { + for item in 0..body["items"].as_array().map_or(0, Vec::len) { + paths.push(format!("/items/{item}/payload")); + } + } + _ => {} + } + } + paths +} diff --git a/src/tests/runtime_uploads.rs b/src/tests/runtime_uploads.rs new file mode 100644 index 0000000..f76f9da --- /dev/null +++ b/src/tests/runtime_uploads.rs @@ -0,0 +1,498 @@ +use super::*; + +const UPLOAD: &str = "/api/external-payloads/v1"; +const DISCOVERY: &str = "/api/cluster/info"; +const COMPLETE: &str = "/worker/workflow-tasks/test/complete"; + +fn policy() -> Value { + json!({"limits":{"max_payload_bytes":2048}, "namespace":{"external_payload_storage":{ + "status":"available", "threshold_bytes":64, "transport":{ + "schema":"durable-workflow.v2.runtime-external-payload-transport.v1", "version":1, + "reference_schema":crate::runtime_payloads::SCHEMA, "mode":"authenticated_namespace_runtime", + "upload":{"method":"POST", "path":UPLOAD}, + "fetch":{"method":"GET", "path_template":"/api/external-payloads/v1/{referenceId}"}, + "limits":{"max_payload_bytes":1048576, "request_timeout_seconds":1} + } + }}}) +} + +fn reference(blob: &str) -> Value { + let hash = format!("{:x}", Sha256::digest(blob.as_bytes())); + json!({"schema":crate::runtime_payloads::SCHEMA, "codec":"avro", + "reference_id":format!("ep_{}", hash[..26].to_uppercase()), + "size_bytes":blob.len(), "sha256":hash}) +} + +fn payload() -> Value { + encode_value_envelope(&json!("a".repeat(128)), DEFAULT_CODEC).unwrap() +} + +fn responses(path: &str, blob: &str, number: usize) -> Option<(&'static str, String)> { + if path.ends_with(DISCOVERY) { + let mut value = policy(); + let storage = &mut value["namespace"]["external_payload_storage"]; + match path.split('/').nth(1).unwrap_or("") { + "unavailable" => storage["status"] = json!("unavailable"), + "aggregate" => storage["threshold_bytes"] = json!(1048576), + "boundary" => { + storage["threshold_bytes"] = json!(payload()["blob"].as_str().unwrap().len()) + } + "schema" => storage["transport"]["schema"] = json!("invalid"), + "version" => storage["transport"]["version"] = json!(2), + "threshold" => storage["threshold_bytes"] = json!(0), + "limit" => storage["transport"]["limits"]["max_payload_bytes"] = json!(2), + "timeout" => storage["transport"]["limits"]["request_timeout_seconds"] = json!(0), + "upload-uri" => { + storage["transport"]["upload"]["path"] = json!("https://elsewhere.invalid/steal") + } + "fetch-uri" => { + storage["transport"]["fetch"]["path_template"] = + json!("https://elsewhere.invalid/{referenceId}") + } + "request-limit" => value["limits"]["max_payload_bytes"] = json!(-1), + _ => {} + } + return Some(("200 OK", value.to_string())); + } + if path.ends_with(UPLOAD) { + if path.starts_with("/pressure/") && number <= 2 { + return Some(( + "503 Service Unavailable", + storage_refusal(None, false, false).to_string(), + )); + } + if path.starts_with("/stopped/") { + return Some(( + "503 Service Unavailable", + storage_refusal(None, false, false).to_string(), + )); + } + let mut value = json!({"schema":"durable-workflow.v2.runtime-external-payload-upload.v1", + "transport_version":1, "reference":reference(blob)}); + match path.split('/').nth(1).unwrap_or("") { + "bad-sha" => value["reference"]["sha256"] = json!("0".repeat(64)), + "bad-size" => value["reference"]["size_bytes"] = json!(1), + "bad-codec" => value["reference"]["codec"] = json!("json"), + "bad-id" => value["reference"]["reference_id"] = json!("../other"), + "bad-extra" => value["reference"]["extra"] = json!(true), + "bad-response" => value["transport_version"] = json!(2), + "bad-json" => return Some(("200 OK", "broken".into())), + "huge-response" => return Some(("200 OK", "x".repeat(65537))), + "unauthorized" => return Some(("403 Forbidden", "access denied".into())), + "missing" => { + return Some(( + "404 Not Found", + r#"{"reason":"external_payload_not_found"}"#.into(), + )) + } + "slow" => thread::sleep(Duration::from_secs(2)), + _ => {} + } + return Some(("201 Created", value.to_string())); + } + Some(("200 OK", "{}".into())) +} + +fn server() -> MockWorkerServer { + MockWorkerServer::start_with_behavior(MockWorkerBehavior { + request_override: Some(responses), + ..MockWorkerBehavior::default() + }) +} + +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 runtime_upload_preserves_encoded_types_role_headers_and_base_path() { + let server = server(); + let client = Client::builder(format!("{}/runtime", server.base_url())) + .worker_token(Some("worker-only".into())) + .namespace("tenant-a") + .build() + .unwrap(); + let typed = AvroValue::Map(BTreeMap::from([ + ("long".into(), AvroValue::Long(7)), + ("double".into(), AvroValue::Double(7.0)), + ("zero".into(), AvroValue::Double(-0.0)), + ("bytes".into(), AvroValue::Bytes(vec![0, 255, 1])), + ("text".into(), AvroValue::String("a".repeat(128))), + ])); + let envelope = encode_typed_envelope(&typed, DEFAULT_CODEC).unwrap(); + send( + &client, + COMPLETE, + true, + json!({"commands":[{"type":"complete_workflow", "result":envelope}]}), + ) + .await + .unwrap(); + let requests = server.requests.lock().unwrap(); + let upload = requests + .iter() + .find(|request| request.path.ends_with(UPLOAD)) + .unwrap(); + assert_eq!(upload.body, envelope["blob"].as_str().unwrap()); + assert_eq!(decode_avro_value_blob(&upload.body).unwrap(), typed); + let headers = upload.headers.to_lowercase(); + assert!(headers.contains("content-type: application/octet-stream")); + assert!(headers.contains("x-durable-workflow-payload-codec: avro")); + assert!(headers.contains(&format!( + "x-durable-workflow-payload-size: {}", + upload.body.len() + ))); + assert!(headers.contains(&format!( + "x-durable-workflow-payload-sha256: {:x}", + Sha256::digest(upload.body.as_bytes()) + ))); + for request in requests.iter() { + assert!(request.path.starts_with("/runtime/api/")); + assert_eq!(request.namespace.as_deref(), Some("tenant-a")); + assert_eq!(request.authorization.as_deref(), Some("Bearer worker-only")); + } + let discovery = &requests[0]; + assert_eq!( + discovery.control_protocol.as_deref(), + Some(CONTROL_PLANE_VERSION) + ); + assert!(discovery.worker_protocol.is_none()); + assert_eq!( + upload.worker_protocol.as_deref(), + Some(WORKER_PROTOCOL_VERSION) + ); + assert!(upload.control_protocol.is_none()); + let completed: Value = serde_json::from_str(&requests.last().unwrap().body).unwrap(); + assert_eq!( + completed["commands"][0]["result"]["external_payload"], + reference(&upload.body) + ); +} + +#[tokio::test] +async fn runtime_upload_deduplicates_per_request_but_keeps_role_cache_separate() { + let server = server(); + let client = Client::builder(server.base_url()) + .control_token(Some("client".into())) + .worker_token(Some("worker".into())) + .build() + .unwrap(); + let body = json!({"commands":[ + {"type":"schedule_activity", "arguments":payload()}, + {"type":"complete_workflow", "result":payload()} + ]}); + for _ in 0..2 { + send(&client, COMPLETE, true, body.clone()).await.unwrap(); + } + assert_eq!(server.request_count(UPLOAD), 2); + assert_eq!(server.request_count(DISCOVERY), 1); + assert_identical_requests(&server, UPLOAD, 2); + assert_identical_requests(&server, &format!("/api{COMPLETE}"), 2); + send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap(); + assert_eq!(server.request_count(DISCOVERY), 2); + let requests = server.requests.lock().unwrap(); + let roles: Vec<_> = requests + .iter() + .filter(|r| r.path == DISCOVERY) + .map(|r| r.authorization.as_deref()) + .collect(); + assert_eq!(roles, [Some("Bearer worker"), Some("Bearer client")]); +} + +#[tokio::test] +async fn runtime_upload_covers_only_protocol_payload_positions() { + for (path, worker, body, pointer) in [ + ( + "/workflows", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/activities", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/workflows/w/signal/s", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/workflows/w/runs/r/query/q", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/workflows/w/update/u", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/workflows/w/message-streams/s/messages", + false, + json!({"input":payload()}), + "/input/external_payload", + ), + ( + "/schedules/s", + false, + json!({"action":{"input":payload()}}), + "/action/input/external_payload", + ), + ( + "/service-endpoints/e/services/s/operations/o/execute", + false, + json!({"arguments":payload()}), + "/arguments/external_payload", + ), + ( + "/worker/activity-tasks/t/complete", + true, + json!({"result":payload()}), + "/result/external_payload", + ), + ( + "/worker/activity-tasks/t/fail", + true, + json!({"failure":{"details":payload()}}), + "/failure/details/external_payload", + ), + ( + "/worker/query-tasks/t/complete", + true, + json!({"result":"duplicate".repeat(1024),"result_envelope":payload()}), + "/result_envelope/external_payload", + ), + ( + "/workflows/w/runs/r/streams/s/items", + false, + json!({"items":[{"payload":payload()["blob"], "payload_codec":"avro"}]}), + "/items/0/payload_reference", + ), + ] { + let server = server(); + let client = Client::new(server.base_url()).unwrap(); + send(&client, path, worker, body).await.unwrap(); + let actual = server.request_body(&format!("/api{path}")); + assert_eq!( + actual.pointer(pointer), + Some(&reference(payload()["blob"].as_str().unwrap())), + "{path}" + ); + if pointer.starts_with("/result_envelope") { + assert!(actual["result"].is_null()); + } + if pointer.ends_with("payload_reference") { + assert!(actual["items"][0].get("payload").is_none()); + } + } + for kind in [ + "complete_workflow", + "complete_update", + "record_side_effect", + "schedule_activity", + "start_child_workflow", + "continue_as_new", + "start_service_operation", + "upsert_memo", + ] { + let server = server(); + let client = Client::new(server.base_url()).unwrap(); + let field = workflow_command_payload_field(kind).unwrap(); + send( + &client, + COMPLETE, + true, + json!({"commands":[{"type":kind, field:payload()}], "memo":{"business":payload()}}), + ) + .await + .unwrap(); + let actual = server.request_body(&format!("/api{COMPLETE}")); + assert!(actual["commands"][0][field] + .get("external_payload") + .is_some()); + assert_eq!(actual["memo"]["business"], payload()); + } +} + +#[tokio::test] +async fn runtime_upload_plans_aggregate_limits_before_any_effect() { + let server = server(); + let client = Client::new(format!("{}/aggregate", server.base_url())).unwrap(); + let envelope = encode_value_envelope(&json!("x".repeat(900)), DEFAULT_CODEC).unwrap(); + send(&client, COMPLETE, true, json!({"commands":[ + {"type":"schedule_activity", "arguments":envelope}, {"type":"complete_workflow", "result":envelope} + ]})).await.unwrap(); + assert_eq!(server.request_count(&format!("/aggregate{UPLOAD}")), 1); + let requests = server.requests.lock().unwrap(); + assert!(requests.last().unwrap().body.len() <= 2048); + drop(requests); + for body in [ + json!({"input":payload(), "metadata":"x".repeat(3000)}), + json!({"input":{"codec":"avro","blob":"x".repeat(1048577)}}), + ] { + let before = server.request_count(&format!("/aggregate{UPLOAD}")); + assert!(send(&client, "/workflows", false, body).await.is_err()); + assert_eq!(server.request_count(&format!("/aggregate{UPLOAD}")), before); + } +} + +#[tokio::test] +async fn runtime_upload_obeys_threshold_and_unavailable_storage() { + for prefix in ["boundary", "unavailable"] { + let server = server(); + let client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); + let small = if prefix == "boundary" { + payload() + } else { + encode_value_envelope(&Value::Null, DEFAULT_CODEC).unwrap() + }; + send(&client, "/workflows", false, json!({"input":small})) + .await + .unwrap(); + assert_eq!(server.request_count(&format!("/{prefix}{UPLOAD}")), 0); + if prefix == "unavailable" { + let error = send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap_err(); + assert!(error.to_string().contains("external_payload_unavailable")); + } + } +} + +#[tokio::test] +async fn runtime_upload_rejects_invalid_discovery_before_upload() { + for prefix in [ + "schema", + "version", + "threshold", + "limit", + "timeout", + "upload-uri", + "fetch-uri", + "request-limit", + ] { + let server = server(); + let client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); + let error = send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap_err(); + assert!( + error.to_string().contains("external_payload_unsupported"), + "{prefix}: {error}" + ); + assert_eq!(server.captured_paths(), [format!("/{prefix}{DISCOVERY}")]); + } +} + +#[tokio::test] +async fn runtime_upload_rejects_corrupt_responses_and_preserves_http_errors() { + for (prefix, reason) in [ + ("bad-sha", "external_payload_integrity_mismatch"), + ("bad-size", "external_payload_integrity_mismatch"), + ("bad-codec", "external_payload_unsupported"), + ("bad-id", "external_payload_unsupported"), + ("bad-extra", "external_payload_unsupported"), + ("bad-response", "external_payload_unsupported"), + ("bad-json", "external_payload_unsupported"), + ("huge-response", "external_payload_unsupported"), + ("unauthorized", "http 403"), + ("missing", "external_payload_not_found"), + ("slow", "transport error"), + ] { + let server = server(); + let client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); + let error = send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap_err(); + assert!(error.to_string().contains(reason), "{prefix}: {error}"); + assert_eq!(server.request_count(&format!("/{prefix}/api/workflows")), 0); + } +} + +#[tokio::test] +async fn runtime_upload_storage_admission_reuses_identical_bytes_and_shutdown_interrupts() { + for prefix in ["pressure", "stopped"] { + let server = server(); + let mut client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); + let stop = Arc::new(AtomicBool::new(prefix == "stopped")); + client.worker_storage_admission = Some(WorkerStorageAdmission { + stop, + policy: WorkerRetryPolicy { + max_backoff: Duration::from_millis(1), + ..WorkerRetryPolicy::default() + }, + }); + let outcome = send( + &client, + COMPLETE, + true, + json!({"commands":[{"type":"complete_workflow","result":payload()}]}), + ) + .await; + if prefix == "pressure" { + outcome.unwrap(); + assert_identical_requests(&server, &format!("/{prefix}{UPLOAD}"), 3); + assert_eq!(server.request_count(&format!("/{prefix}/api{COMPLETE}")), 1); + } else { + assert!(worker_storage_admission_body(&outcome.unwrap_err()).is_some()); + assert_eq!(server.request_count(&format!("/{prefix}{UPLOAD}")), 1); + assert_eq!(server.request_count(&format!("/{prefix}/api{COMPLETE}")), 0); + } + } +} + +#[tokio::test] +async fn runtime_upload_low_level_references_are_strict_and_never_reuploaded() { + let server = server(); + let client = Client::new(server.base_url()).unwrap(); + let envelope = + json!({"codec":"avro","external_payload":reference(payload()["blob"].as_str().unwrap())}); + client + .complete_workflow_task( + "test", + "worker", + 1, + vec![json!({"type":"complete_workflow","result":envelope})], + ) + .await + .unwrap(); + assert_eq!(server.request_count(UPLOAD), 0); + assert_eq!(server.request_count(DISCOVERY), 0); + for (path, bad) in [ + ("/codec", json!("json")), + ("/external_payload/reference_id", json!("../../other")), + ("/external_payload/size_bytes", json!(-1)), + ] { + let mut value = envelope.clone(); + *value.pointer_mut(path).unwrap() = bad; + assert!(client + .complete_workflow_task( + "test", + "worker", + 1, + vec![json!({"type":"complete_workflow","result":value})] + ) + .await + .is_err()); + } + assert_eq!(server.request_count(&format!("/api{COMPLETE}")), 1); +} From 8b6632b0c809ac8b60669153057edd9f093b89fd Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 11:32:25 +0000 Subject: [PATCH 2/2] Qualify Rust runtime uploads against native Server and cold restart --- CHANGELOG.md | 10 + Cargo.toml | 4 +- README.md | 4 +- examples/runtime_payloads.rs | 303 ++++++++++++++++++++++++++++ scripts/ci/test-publish-rust-sdk.py | 2 +- src/lib.rs | 4 +- src/runtime_uploads.rs | 11 +- src/tests/runtime_uploads.rs | 169 +++++++++++++++- tests/replay_regression_corpus.rs | 1 + tests/runtime-payloads.compose.yml | 75 +++++++ tests/runtime-payloads.md | 57 ++++++ 11 files changed, 627 insertions(+), 13 deletions(-) create mode 100644 examples/runtime_payloads.rs create mode 100644 tests/runtime-payloads.compose.yml create mode 100644 tests/runtime-payloads.md diff --git a/CHANGELOG.md b/CHANGELOG.md index afcaa02..35cf4f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## 2.0.4 + +- Upload large encoded payloads through the authenticated namespace runtime + before sending client requests or worker commands. Discover runtime limits, + externalize aggregate requests when necessary, and preserve exact Avro types. +- Verify upload references, keep discovery and responses bounded, and reuse + existing worker storage-admission waits without re-executing handlers. +- Include a native runtime qualification for large typed input, activity and + workflow results, query/signal recovery, maximum size and cold restart. + ## 2.0.3 - Resolve Server-managed external payload references before decoding worker diff --git a/Cargo.toml b/Cargo.toml index 616526c..21e9f31 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "durable-workflow" -version = "2.0.3" +version = "2.0.4" 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.3" +product-train = "2.0.4" compatibility-authority = "protocol-manifests" supported-server-versions = "2.0.0" qualified-server-version = "2.0.0" diff --git a/README.md b/README.md index 2be01b0..49820ce 100644 --- a/README.md +++ b/README.md @@ -81,7 +81,9 @@ nulls, booleans, signed 64-bit integers, finite doubles, bytes, UTF-8 strings, lists, and string-keyed maps across official SDKs without customer-managed schemas or a registry. -Server-managed external payloads are fetched automatically through the same +Large payloads are uploaded automatically when the namespace advertises runtime +storage. The SDK follows its inline threshold and upload limit, including batches +that exceed the ordinary request limit. Uploads and downloads use the same runtime URL, namespace, and credential role, with size and SHA-256 verification. `Client::builder(...).max_external_payload_bytes(...)` limits unique downloaded bytes per response (64 MiB by default). Provider credentials are not needed; diff --git a/examples/runtime_payloads.rs b/examples/runtime_payloads.rs new file mode 100644 index 0000000..452665d --- /dev/null +++ b/examples/runtime_payloads.rs @@ -0,0 +1,303 @@ +//! Native external-payload qualification against an isolated Server. +//! See tests/runtime-payloads.md. Never point this fixture at a customer runtime. +use durable_workflow::{ + decode_avro_value, encode_payload, json, wait_condition, ActivityOptions, AvroValue, Client, + PayloadEnvelope, Value, Worker, WorkflowDescription, +}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{collections::BTreeMap, time::Duration}; + +type Result = std::result::Result>; +const NS: &str = "external-proof"; +const WORKFLOW: &str = "external-payload.rust"; +const ACTIVITY: &str = "external-payload.rust.echo"; +const MAXIMUM: usize = 50_331_633; + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +struct Payload { + text: String, + long: i64, + double: f64, + negative_zero: f64, + #[serde(with = "serde_bytes")] + binary: Vec, + nested: BTreeMap>, +} + +#[derive(Deserialize, Serialize)] +struct Request { + value: Payload, + wait: bool, +} + +fn payload() -> Payload { + Payload { + text: "durable-external-value-".repeat(131072), + long: 7, + double: 7.0, + negative_zero: -0.0, + binary: vec![0, 255, 128, 1], + nested: BTreeMap::from([( + "nested".into(), + BTreeMap::from([("z".into(), 2), ("a".into(), 1)]), + )]), + } +} + +fn assert_payload(value: AvroValue) -> Result<()> { + // Compare on the wire before Serde conversion: 7 and 7.0 are not the same value. + let expected = decode_avro_value(&encode_payload(&payload(), "avro")?)?; + assert_eq!(value, expected, "lossless Avro value changed"); + let actual: Payload = value.deserialize()?; + assert_eq!(actual, payload()); + assert!(actual.negative_zero.is_sign_negative()); + Ok(()) +} + +fn token(role: &str) -> String { + format!("{:x}", Sha256::digest(format!("external-proof-{role}"))) +} + +fn client(url: &str, worker: bool) -> Result { + let builder = Client::builder(url) + .namespace(NS) + .timeout(Duration::from_secs(120)); + Ok(if worker { + builder.worker_token(Some(token("worker"))) + } else { + builder.control_token(Some(token("operator"))) + } + .build()?) +} + +async fn api(url: &str, method: &str, path: &str, body: Value) -> Result { + let response = reqwest::Client::new() + .request(method.parse()?, format!("{url}/api{path}")) + .bearer_auth("external-payload-fixture") + .header("X-Namespace", NS) + .header("X-Durable-Workflow-Control-Plane-Version", "2") + .json(&body) + .send() + .await?; + let status = response.status(); + let value: Value = response.json().await?; + if !status.is_success() { + return Err(format!("fixture API {path}: HTTP {status}: {value}").into()); + } + Ok(value) +} + +async fn wait_for(client: &Client, id: &str, status: &str) -> Result { + let deadline = tokio::time::Instant::now() + Duration::from_secs(120); + loop { + let description = client.describe_workflow(id).await?; + if description.status.as_deref() == Some(status) { + return Ok(description); + } + if description.is_terminal() || tokio::time::Instant::now() >= deadline { + return Err(format!( + "{id}: expected {status}, got {:?}: {:?}", + description.status, description.failure + ) + .into()); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +async fn worker(url: &str) -> Result<()> { + let mut worker = Worker::new(client(url, true)?, NS) + .worker_id("external-rust-worker") + .poll_timeout(Duration::from_secs(1)); + worker.register_typed_activity(ACTIVITY, |ctx, input: Payload| async move { + ctx.heartbeat(json!({"bytes": input.text.len()})).await?; + Ok(input) + }); + worker.register_typed_workflow(WORKFLOW, |ctx, input: Request| async move { + let result: Payload = ctx + .activity_typed_with_options( + ACTIVITY, + ActivityOptions::new().start_to_close_timeout(Duration::from_secs(60)), + input.value, + ) + .await?; + if input.wait { + let predicate_ctx = ctx.clone(); + wait_condition!(ctx, "released", move || Ok(!predicate_ctx + .signals("release")? + .is_empty())) + .await?; + assert_eq!( + ctx.signals("release")?[0], + vec![json!(format!("{:x}", Sha256::digest(&result.text)))] + ); + } + Ok(result) + }); + worker.register_query_avro_value(WORKFLOW, "value", |ctx, _| async move { + let result = ctx + .history_events() + .iter() + .find(|event| event.event_type == "ActivityCompleted") + .ok_or_else(|| { + durable_workflow::Error::WorkerLoop("missing completed activity history".into()) + })?; + let envelope: PayloadEnvelope = serde_json::from_value(result.payload["result"].clone())?; + decode_avro_value(&envelope) + }); + worker.register_typed_workflow("external-payload.rust.maximum", |_, _: ()| async { + Ok("m".repeat(MAXIMUM)) + }); + worker.run().await?; + Ok(()) +} + +#[tokio::main] +async fn main() -> Result<()> { + let url = std::env::var("RUNTIME_URL").unwrap_or_else(|_| "http://server:8080".into()); + assert!( + matches!( + reqwest::Url::parse(&url)?.host_str(), + Some("server" | "localhost" | "127.0.0.1") + ), + "local fixture only" + ); + let phase = std::env::args() + .nth(1) + .ok_or("choose prepare, worker, start, verify, maximum or verify-maximum")?; + if phase == "worker" { + return worker(&url).await; + } + let client = client(&url, false)?; + match phase.as_str() { + "prepare" => { + api( + &url, + "POST", + "/namespaces", + json!({"name":NS,"retention_days":30}), + ) + .await?; + api(&url, "PUT", &format!("/namespaces/{NS}/external-storage"), json!({ + "enabled":true,"driver":"local","threshold_bytes":64,"config":{"uri":"file:///payloads"} + })).await?; + for role in ["operator", "worker"] { + api(&url, "PUT", &format!("/runtime-credentials/external-proof-{role}"), json!({ + "token":token(role),"subject":format!("external-proof-{role}"),"roles":[role],"tenant":NS + })).await?; + } + let info = client.cluster_info().await?; + assert_eq!(info["limits"]["max_payload_bytes"], 2097152); + assert_eq!( + info["namespace"]["external_payload_storage"]["transport"]["limits"] + ["max_payload_bytes"], + 67108864 + ); + } + "start" => { + for (kind, wait) in [("completed", false), ("waiting", true)] { + let id = format!("external-rust-{kind}"); + let handle = client + .start_workflow( + WORKFLOW, + NS, + &id, + Request { + value: payload(), + wait, + }, + ) + .await?; + let description = + wait_for(&client, &id, if wait { "waiting" } else { "completed" }).await?; + if !wait { + assert_payload( + description + .output_avro_value + .ok_or("missing typed output")?, + )?; + } else { + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + loop { + let history = api( + &url, + "GET", + &format!( + "/workflows/{id}/runs/{}/history", + handle.run_id.as_deref().unwrap() + ), + Value::Null, + ) + .await?; + let events = history["events"].as_array().ok_or("missing history")?; + if events + .iter() + .any(|event| event["event_type"] == "ConditionWaitOpened") + { + assert_eq!( + events + .iter() + .filter(|event| event["event_type"] == "ActivityCompleted") + .count(), + 1 + ); + break; + } + if tokio::time::Instant::now() >= deadline { + return Err("workflow never opened its durable condition".into()); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + println!("{kind}: workflow_id={id} run_id={}", handle.run_id.unwrap()); + } + } + "verify" => { + assert_payload( + client + .query_workflow_avro_value("external-rust-waiting", "value", json!([])) + .await?, + )?; + client + .signal_workflow( + "external-rust-waiting", + "release", + [format!("{:x}", Sha256::digest(payload().text))], + ) + .await?; + for kind in ["completed", "waiting"] { + let description = + wait_for(&client, &format!("external-rust-{kind}"), "completed").await?; + assert_payload( + description + .output_avro_value + .ok_or("missing typed output")?, + )?; + } + } + "maximum" | "verify-maximum" => { + if phase == "maximum" { + client + .start_workflow( + "external-payload.rust.maximum", + NS, + "external-rust-maximum", + (), + ) + .await?; + } + let description = wait_for(&client, "external-rust-maximum", "completed").await?; + let actual: String = description + .output_avro_value + .ok_or("missing maximum output")? + .deserialize()?; + assert_eq!(actual.len(), MAXIMUM); + assert!(actual.bytes().all(|byte| byte == b'm')); + assert_eq!(encode_payload(&actual, "avro")?.blob.len(), 67_108_864); + } + _ => return Err("unknown phase".into()), + } + println!("{phase}: passed"); + Ok(()) +} diff --git a/scripts/ci/test-publish-rust-sdk.py b/scripts/ci/test-publish-rust-sdk.py index 48e640a..907318b 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.3" +PACKAGE_VERSION = "2.0.4" PRODUCT_TRAIN = PACKAGE_VERSION SERVER_VERSIONS = "2.0.0" QUALIFIED_SERVER_VERSION = "2.0.0" diff --git a/src/lib.rs b/src/lib.rs index 4d1add3..c283e37 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24371,10 +24371,12 @@ mod tests { thread: Option>, } + type RequestOverride = fn(&str, &str, usize) -> Option<(&'static str, String)>; + #[derive(Clone, Copy, Default)] struct MockWorkerBehavior { response_override: Option Option<(&'static str, String)>>, - request_override: Option Option<(&'static str, String)>>, + request_override: Option, storage_refusals: usize, storage_path: Option<&'static str>, storage_unavailable: bool, diff --git a/src/runtime_uploads.rs b/src/runtime_uploads.rs index dd53459..d0e76eb 100644 --- a/src/runtime_uploads.rs +++ b/src/runtime_uploads.rs @@ -192,7 +192,7 @@ impl Client { .runtime_payload_request(reqwest::Method::GET, "/cluster/info", protocol, true)? .build()?; let info = self - .runtime_payload_json(request, protocol, 512 * 1024) + .runtime_payload_json(request, protocol, 2 * 1024 * 1024) .await?; let policy = Policy::from_info(&info)?; self.runtime_upload_policy @@ -264,10 +264,11 @@ impl Client { bytes.extend_from_slice(&chunk); } if !status.is_success() { - let error = Error::Http { - status, - body: String::from_utf8_lossy(&bytes).into_owned(), - }; + let body = String::from_utf8_lossy(&bytes).into_owned(); + if let Some(failure) = protocol_failure(status, &body) { + return Err(Error::Protocol(failure)); + } + let error = Error::Http { status, body }; if self .wait_for_storage_admission(&error, protocol, None, &mut retries) .await diff --git a/src/tests/runtime_uploads.rs b/src/tests/runtime_uploads.rs index f76f9da..381af98 100644 --- a/src/tests/runtime_uploads.rs +++ b/src/tests/runtime_uploads.rs @@ -50,6 +50,8 @@ fn responses(path: &str, blob: &str, number: usize) -> Option<(&'static str, Str json!("https://elsewhere.invalid/{referenceId}") } "request-limit" => value["limits"]["max_payload_bytes"] = json!(-1), + "full-manifest" => value["worker_protocol"] = json!("x".repeat(600_000)), + "oversized-manifest" => value["worker_protocol"] = json!("x".repeat(2_097_153)), _ => {} } return Some(("200 OK", value.to_string())); @@ -79,6 +81,14 @@ fn responses(path: &str, blob: &str, number: usize) -> Option<(&'static str, Str "bad-json" => return Some(("200 OK", "broken".into())), "huge-response" => return Some(("200 OK", "x".repeat(65537))), "unauthorized" => return Some(("403 Forbidden", "access denied".into())), + "protocol" => return Some(( + "400 Bad Request", + json!({ + "reason":"unsupported_protocol_version", "message":"worker protocol rejected", + "supported_version":"1.19", "requested_version":"1.0" + }) + .to_string(), + )), "missing" => { return Some(( "404 Not Found", @@ -212,6 +222,78 @@ async fn runtime_upload_deduplicates_per_request_but_keeps_role_cache_separate() assert_eq!(roles, [Some("Bearer worker"), Some("Bearer client")]); } +#[tokio::test] +async fn runtime_upload_refreshes_expired_policy_without_sharing_namespaces() { + let server = server(); + let client = Client::builder(server.base_url()) + .namespace("a") + .build() + .unwrap(); + send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap(); + client.runtime_upload_policy.lock().unwrap()[0] + .as_mut() + .unwrap() + .0 = Instant::now() - Duration::from_secs(61); + send( + &client.clone(), + "/workflows", + false, + json!({"input":payload()}), + ) + .await + .unwrap(); + let other = Client::builder(server.base_url()) + .namespace("b") + .build() + .unwrap(); + send(&other, "/workflows", false, json!({"input":payload()})) + .await + .unwrap(); + let requests = server.requests.lock().unwrap(); + let namespaces: Vec<_> = requests + .iter() + .filter(|r| r.path == DISCOVERY) + .map(|r| r.namespace.as_deref()) + .collect(); + assert_eq!(namespaces, [Some("a"), Some("a"), Some("b")]); +} + +#[tokio::test] +async fn runtime_upload_recovery_retains_the_already_computed_activity_outcome() { + fn upload_pressure(path: &str, body: &str, number: usize) -> Option<(&'static str, String)> { + if path == UPLOAD || path == DISCOVERY { + responses(&format!("/pressure{path}"), body, number) + } else { + None + } + } + let server = MockWorkerServer::start_with_behavior(MockWorkerBehavior { + request_override: Some(upload_pressure), + storage_activity: true, + ..MockWorkerBehavior::default() + }); + let mut worker = storage_worker(&server); + let calls = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&calls); + worker.register_activity("storage.activity", move |_, _| { + observed.fetch_add(1, Ordering::SeqCst); + async { Ok(json!("a".repeat(128))) } + }); + assert_eq!(worker.run_once().await.unwrap(), 1); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_identical_requests(&server, UPLOAD, 3); + assert_eq!( + server.request_count("/api/worker/activity-tasks/storage-activity/complete"), + 1 + ); + assert_eq!( + server.request_count("/api/worker/activity-tasks/storage-activity/fail"), + 0 + ); +} + #[tokio::test] async fn runtime_upload_covers_only_protocol_payload_positions() { for (path, worker, body, pointer) in [ @@ -342,9 +424,10 @@ async fn runtime_upload_plans_aggregate_limits_before_any_effect() { {"type":"schedule_activity", "arguments":envelope}, {"type":"complete_workflow", "result":envelope} ]})).await.unwrap(); assert_eq!(server.request_count(&format!("/aggregate{UPLOAD}")), 1); - let requests = server.requests.lock().unwrap(); - assert!(requests.last().unwrap().body.len() <= 2048); - drop(requests); + { + let requests = server.requests.lock().unwrap(); + assert!(requests.last().unwrap().body.len() <= 2048); + } for body in [ json!({"input":payload(), "metadata":"x".repeat(3000)}), json!({"input":{"codec":"avro","blob":"x".repeat(1048577)}}), @@ -389,6 +472,7 @@ async fn runtime_upload_rejects_invalid_discovery_before_upload() { "upload-uri", "fetch-uri", "request-limit", + "oversized-manifest", ] { let server = server(); let client = Client::new(format!("{}/{prefix}", server.base_url())).unwrap(); @@ -403,6 +487,16 @@ async fn runtime_upload_rejects_invalid_discovery_before_upload() { } } +#[tokio::test] +async fn runtime_upload_discovery_accepts_the_full_server_protocol_manifest() { + let server = server(); + let client = Client::new(format!("{}/full-manifest", server.base_url())).unwrap(); + send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap(); + assert_eq!(server.request_count(&format!("/full-manifest{UPLOAD}")), 1); +} + #[tokio::test] async fn runtime_upload_rejects_corrupt_responses_and_preserves_http_errors() { for (prefix, reason) in [ @@ -415,6 +509,7 @@ async fn runtime_upload_rejects_corrupt_responses_and_preserves_http_errors() { ("bad-json", "external_payload_unsupported"), ("huge-response", "external_payload_unsupported"), ("unauthorized", "http 403"), + ("protocol", "protocol rejected"), ("missing", "external_payload_not_found"), ("slow", "transport error"), ] { @@ -428,6 +523,74 @@ async fn runtime_upload_rejects_corrupt_responses_and_preserves_http_errors() { } } +#[tokio::test] +async fn runtime_upload_bounds_chunked_responses_and_does_not_follow_location() { + for redirect in [false, true] { + let target = TcpListener::bind("127.0.0.1:0").unwrap(); + target.set_nonblocking(true).unwrap(); + let location = format!("http://{}/credential-sink", target.local_addr().unwrap()); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let transport = thread::spawn(move || { + for discovery in [true, false] { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut request = Vec::new(); + loop { + let mut buffer = [0; 4096]; + let read = stream.read(&mut buffer).unwrap(); + assert!(read > 0); + request.extend_from_slice(&buffer[..read]); + if mock_request_is_complete(&request) { + break; + } + } + if discovery { + write_mock_response(&mut stream, "200 OK", &policy().to_string()); + } else { + let response = if redirect { + format!("HTTP/1.1 307 Temporary Redirect\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + } else { + format!("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n10001\r\n{}\r\n0\r\n\r\n", "x".repeat(65537)) + }; + let _ = stream.write_all(response.as_bytes()); + } + } + }); + let client = Client::builder(format!("http://{address}")) + .control_token(Some("client-fixture".into())) + .timeout(Duration::from_secs(2)) + .build() + .unwrap(); + let error = send(&client, "/workflows", false, json!({"input":payload()})) + .await + .unwrap_err(); + if redirect { + assert!(matches!( + error, + Error::Http { + status: reqwest::StatusCode::TEMPORARY_REDIRECT, + .. + } + )); + } else { + assert!( + error + .to_string() + .contains("response exceeds its byte limit"), + "{error}" + ); + } + assert_eq!( + target.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + transport.join().unwrap(); + } +} + #[tokio::test] async fn runtime_upload_storage_admission_reuses_identical_bytes_and_shutdown_interrupts() { for prefix in ["pressure", "stopped"] { diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index d8e6804..ab35319 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -171,6 +171,7 @@ fn handle_request( && path.ends_with("/complete") && request_number as u64 <= completion_storage_refusals; let body = match path.as_str() { + "/api/cluster/info" => json!({"limits": {"max_payload_bytes": 2097152}}).to_string(), _ if storage_refused => json!({ "reason": "storage_pressure", "storage_state": "fenced", diff --git a/tests/runtime-payloads.compose.yml b/tests/runtime-payloads.compose.yml new file mode 100644 index 0000000..a85fdd0 --- /dev/null +++ b/tests/runtime-payloads.compose.yml @@ -0,0 +1,75 @@ +name: rust-payload-proof +x-runtime: &runtime + image: ${SERVER_IMAGE:?Set an exact published Server image} + user: "1000:1000" + environment: &environment + APP_ENV: testing + APP_DEBUG: "false" + DW_SERVER_KEY: "base64:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" + DB_CONNECTION: mysql + DB_HOST: mysql + DB_DATABASE: durable_workflow + DB_USERNAME: workflow + DB_PASSWORD: workflow + CACHE_STORE: redis + REDIS_HOST: redis + QUEUE_CONNECTION: redis + DW_AUTH_TOKEN: external-payload-fixture + DW_RUNTIME_CREDENTIALS_ENABLED: "true" + DW_AUTH_BACKWARD_COMPATIBLE: "true" + volumes: + - ../target/runtime-payloads:/payloads +services: + mysql: + image: mysql:8.0.46 + environment: + MYSQL_DATABASE: durable_workflow + MYSQL_USER: workflow + MYSQL_PASSWORD: workflow + MYSQL_ROOT_PASSWORD: fixture-root + volumes: + - mysql:/var/lib/mysql + mem_limit: 768m + healthcheck: + test: [CMD, mysqladmin, ping, -h, 127.0.0.1, -pfixture-root] + interval: 2s + timeout: 2s + retries: 45 + redis: + image: redis:7-alpine + mem_limit: 128m + bootstrap: + <<: *runtime + command: [server-bootstrap] + healthcheck: + disable: true + depends_on: + mysql: + condition: service_healthy + server: + <<: *runtime + mem_limit: 256m + cpus: 1 + depends_on: + bootstrap: + condition: service_completed_successfully + redis: + condition: service_started + healthcheck: + test: [CMD, server-healthcheck] + interval: 3s + timeout: 3s + retries: 30 + sdk: + image: rust:1.86-bookworm + user: "1000:1000" + working_dir: /sdk + entrypoint: [/sdk/target/debug/examples/runtime_payloads] + environment: + RUNTIME_URL: http://server:8080 + volumes: + - ..:/sdk:ro + mem_limit: 2g + profiles: [client] +volumes: + mysql: diff --git a/tests/runtime-payloads.md b/tests/runtime-payloads.md new file mode 100644 index 0000000..dc5cab6 --- /dev/null +++ b/tests/runtime-payloads.md @@ -0,0 +1,57 @@ +# Native Runtime Payload Qualification + +This is a destructive, local-only experiment. Do not use customer namespaces or +production credentials. The fixture creates its own namespace and role-specific +test credentials. It uses the native published Server, MySQL and Redis; it does +not replace the HTTP server or lift ordinary request limits. + +Set `SERVER_IMAGE` to an exact published image compatible with runtime external +payload transport. Record its digest and this checkout SHA in the PR or release +issue. For publication qualification, run the same example in a clean Cargo +consumer with an exact registry dependency instead of a path override. + +From the repository root (Rust 1.86+, Docker Compose): + +```sh +cargo build --example runtime_payloads +mkdir -p target/runtime-payloads +docker compose -f tests/runtime-payloads.compose.yml up -d --wait --wait-timeout 180 server +docker compose -f tests/runtime-payloads.compose.yml run --rm sdk prepare +docker compose -f tests/runtime-payloads.compose.yml run -d --name rust-payload-worker sdk worker +docker compose -f tests/runtime-payloads.compose.yml run --rm sdk start +docker compose -f tests/runtime-payloads.compose.yml run --rm sdk maximum +``` + +`start` must complete one workflow and persist a second workflow after its +activity reaches a condition wait. The payload contains 3,014,656 text bytes, +binary bytes including invalid UTF-8, an integer, an integral double, signed +zero and a nested map. `maximum` verifies an exactly 64 MiB encoded result. +No payload contents are printed. + +Stop the worker, Server and MySQL; keep the database volume and payload files. +The worker uses abrupt termination deliberately to exercise process loss. + +```sh +docker stop --timeout 2 rust-payload-worker +docker compose -f tests/runtime-payloads.compose.yml stop server mysql +docker compose -f tests/runtime-payloads.compose.yml up -d --wait --wait-timeout 180 server +docker start rust-payload-worker +docker compose -f tests/runtime-payloads.compose.yml run --rm sdk verify +docker compose -f tests/runtime-payloads.compose.yml run --rm sdk verify-maximum +``` + +`verify` fetches the committed activity through a cold query snapshot, sends a +digest signal, completes the waiting workflow and compares both results on the +lossless Avro surface before decoding into Serde types. `verify-maximum` fetches +the maximum result through a fresh client after the database/runtime restart. +Report failures as failures, including OOM or timeouts. Record memory limits, +container memory peaks and OOM flags alongside the outcome. This experiment +does not qualify a production storage provider or multi-tenant capacity. + +Cleanup only these fixture resources: + +```sh +docker rm -f rust-payload-worker +docker compose -f tests/runtime-payloads.compose.yml down --volumes --remove-orphans +rm -r target/runtime-payloads +```