From 83252cd92239666d20cad5104dc9f6527166b52e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 04:27:18 +0000 Subject: [PATCH 1/2] Return pending updates as UpdateStillPending and read outcomes later by id (#212) update_workflow returned the Server's result: null for a 202 wait_timed_out receipt, so an update still running looked like a handler that returned null. A pending receipt now returns Error::UpdateStillPending with the update id, request id, current run and full receipt; a real null result is still Ok(Null). Client::update_outcome, update_outcome_avro_value and update_outcome_receipt (and WorkflowHandle::update_outcome) call GET /workflows/{id}/updates/{update_id} (server#1131) on a Server that advertises update_outcome_reads. Without the capability they return Error::UpdateOutcomeReadsUnavailable before reading anything and never resubmit the update. Error::is_update_not_found detects 404 update_not_found. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 19 ++ README.md | 42 ++++ src/lib.rs | 58 ++++-- src/metrics/routes.rs | 1 + src/update_outcome.rs | 368 ++++++++++++++++++++++++++++++++++ src/update_outcome/tests.rs | 383 ++++++++++++++++++++++++++++++++++++ 6 files changed, 854 insertions(+), 17 deletions(-) create mode 100644 src/update_outcome.rs create mode 100644 src/update_outcome/tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 87bd8727..42cf69e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,25 @@ ## Unreleased +- An update whose wait ends before its handler finishes now returns + `Error::UpdateStillPending` instead of `Ok(Null)` (durable-workflow/.github#212). + `update_workflow` used to return the Server's `result: null` for a 202 + `wait_timed_out` receipt, so a pending update looked like a handler that + returned null. The error carries the `update_id`, `request_id`, current + `run_id`, `update_status` and full receipt; a real null result is still + `Ok(Null)`. `update_workflow_avro_value` returns the same error instead of a + `missing_payload_envelope` codec error. + +- `Client::update_outcome`, `Client::update_outcome_avro_value` and + `Client::update_outcome_receipt` (and `WorkflowHandle::update_outcome`) read an + accepted update's outcome later by its update id, without submitting it + again (server#1131). They need a Server that advertises + `update_outcome_reads` (native 3.6.0 and later) and return + `Error::UpdateOutcomeReadsUnavailable` otherwise, before reading anything. A + still-running update returns `Error::UpdateStillPending` with a hint to read + again by update id. A failed update is `Error::Http` 422, and + `Error::is_update_not_found()` detects 404 `update_not_found`. + - A `request_id_conflict` refusal from native Server 3.9.0 (server#934, server#1191) now names the key that was sent. Native doesn't echo the key, so `Error::request_id()` returned `None` and `RequestIdConflict::request_id` was diff --git a/README.md b/README.md index d544c712..2ba4051b 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,48 @@ Older Servers (PHP Server 2.x and native up to 3.8) don't compare input under a key: a repeated signal key records a second signal, and an update retried with changed arguments returns the original receipt. +### Pending updates and reading an outcome later + +`update_workflow` waits for the update to complete. If the Server's wait ends +first, it answers HTTP 202 with no result, and the SDK returns +`Error::UpdateStillPending` with the `update_id`, `request_id`, current +`run_id` and full receipt. A handler that returned null is still `Ok(Null)`. + +On a Server that advertises `capabilities.update_outcome_reads` in +`Client::cluster_info()` (native 3.6.0 and later), +`Client::update_outcome(workflow_id, update_id, wait_timeout_seconds)` reads +the same update by its id without submitting it again. It waits for the +outcome and returns the decoded result, or `Error::UpdateStillPending` again +while the update is still running. A failed update is `Error::Http` with status +422, and an unknown or pruned update id is 404 `update_not_found` +(`Error::is_update_not_found()`), which does not prove the update was never +admitted. `update_outcome_receipt` returns the raw receipt and can answer at +once with `UpdateWaitStage::Accepted`. + +On any other Server, `update_outcome` returns +`Error::UpdateOutcomeReadsUnavailable` before reading anything; the SDK never +resubmits an update in place of reading it. There, call `update_workflow` again +with the same `request_id` to keep waiting. + +```rust,no_run +# async fn rename(client: &durable_workflow::Client) -> durable_workflow::Result<()> { +use durable_workflow::Error; + +let result = match client + .update_workflow("order-1", "rename", ["Grace"], Some("rename-order-1")) + .await +{ + Err(Error::UpdateStillPending(pending)) => { + let update_id = pending.update_id.expect("Server receipt names the update"); + client.update_outcome("order-1", &update_id, Some(30)).await? + } + other => other?, +}; +# let _ = result; +# Ok(()) +# } +``` + ### Condition predicate identities `wait_condition!` derives the recorded predicate identity from the call site's diff --git a/src/lib.rs b/src/lib.rs index e2592e26..a77ae82f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,6 +32,7 @@ pub use pagination::{ }; mod terminal_failure_detail; mod tls; +mod update_outcome; mod worker_capacity; mod worker_session; mod workflow_hydration; @@ -91,6 +92,7 @@ pub use serde_json::{json, Value}; use sha2::{Digest, Sha256}; pub use sticky_workflow_cache::{StickyCacheMetrics, StickyCacheOptions}; use thiserror::Error; +pub use update_outcome::{UpdatePending, UpdateWaitStage}; pub use uuid::Uuid; pub use worker_session::{WorkerSession, WorkerSessionOptions}; @@ -408,6 +410,16 @@ pub enum Error { "workflow_memo_updates_unavailable: the connected runtime did not advertise workflow memo update support" )] WorkflowMemoUpdatesUnavailable, + /// An accepted update had not completed when the wait ended; it has no + /// result yet. See [`UpdatePending`]. + #[error(transparent)] + UpdateStillPending(UpdatePending), + /// The connected Server does not advertise `update_outcome_reads`, so an + /// update outcome cannot be read by update id. Nothing was read or sent. + #[error( + "update_outcome_reads_unavailable: the connected Server does not advertise update_outcome_reads, so an update outcome cannot be read by update id; call update_workflow() again with the same request_id instead" + )] + UpdateOutcomeReadsUnavailable, #[error("cooperative cancellation is unavailable: {0}")] CooperativeCancellationUnavailable(String), #[error("cancellation_scope_execution_not_supported: Rust worker cannot replay scoped cancellation history")] @@ -498,6 +510,24 @@ impl Error { } } + /// Whether an update outcome read found no retained update with that id + /// for the workflow (HTTP 404, `reason: "update_not_found"`). + /// + /// That includes an update pruned with its run, so it does not prove the + /// update was never admitted. + pub fn is_update_not_found(&self) -> bool { + matches!(self, Self::Http { status, .. } if status.as_u16() == 404) + && self + .refusal_body() + .and_then(|body| { + body.get("reason") + .and_then(Value::as_str) + .map(str::to_string) + }) + .as_deref() + == Some("update_not_found") + } + fn refusal_body(&self) -> Option { match self { Self::Http { body, .. } => serde_json::from_str(body).ok(), @@ -3096,6 +3126,8 @@ pub struct Client { activity_lease_renewal_supported: Arc, /// Server #1017: the Server sends sticky hits only the history after the cached sequence. sticky_history_suffix_supported: Arc, + /// Server #1131: the Server advertised `update_outcome_reads`. + update_outcome_reads_supported: Arc, activity_lease_discovery: Arc, activity_lease_authority: Option>, #[cfg(feature = "telemetry")] @@ -3689,6 +3721,12 @@ impl Client { /// key for the same update intent returns the original update's receipt and /// result instead of admitting a second update; a different intent is /// refused with [`Error::is_request_id_conflict`]. + /// + /// The Server waits for the update to complete. When its wait ends first, + /// this returns [`Error::UpdateStillPending`], never `Null`: read the + /// outcome later with [`Client::update_outcome`] and the pending update's + /// id, or send the update again with the same `request_id`. A handler that + /// returned null is `Ok(Value::Null)`. pub async fn update_workflow( &self, workflow_id: &str, @@ -3699,13 +3737,7 @@ impl Client { let response = self .update_workflow_response(workflow_id, update_name, input, request_id) .await?; - if let Some(envelope) = response - .get("result_envelope") - .filter(|envelope| !envelope.is_null()) - { - return decode_wire_value(envelope, DEFAULT_CODEC); - } - Ok(response.get("result").cloned().unwrap_or(response)) + Self::submitted_update_value(workflow_id, response) } /// Send a synchronous update and retain a bytes-capable Avro result. @@ -3719,16 +3751,7 @@ impl Client { let response = self .update_workflow_response(workflow_id, update_name, input, request_id) .await?; - let envelope = response - .get("result_envelope") - .filter(|envelope| !envelope.is_null()) - .ok_or_else(|| { - Error::Codec( - "missing_payload_envelope: typed update result requires result_envelope" - .to_string(), - ) - })?; - decode_wire_avro_value(envelope, DEFAULT_CODEC) + Self::submitted_update_avro_value(workflow_id, response) } async fn update_workflow_response( @@ -6131,6 +6154,7 @@ impl ClientBuilder { failure_request_limit: Arc::new(AtomicUsize::new(failure_reporting::REQUEST_BYTES)), activity_lease_renewal_supported: Arc::new(AtomicBool::new(false)), sticky_history_suffix_supported: Arc::new(AtomicBool::new(false)), + update_outcome_reads_supported: Arc::new(AtomicBool::new(false)), activity_lease_discovery: Arc::new(activity_lease::Discovery::default()), activity_lease_authority: None, #[cfg(feature = "telemetry")] diff --git a/src/metrics/routes.rs b/src/metrics/routes.rs index bc3ab5b5..db445b0a 100644 --- a/src/metrics/routes.rs +++ b/src/metrics/routes.rs @@ -73,6 +73,7 @@ const ROUTES: &[&str] = &[ "/workflows/{workflow_id}/signal/{name}", "/workflows/{workflow_id}/query/{name}", "/workflows/{workflow_id}/update/{name}", + "/workflows/{workflow_id}/updates/{update_id}", "/workflows/{workflow_id}/message-streams/{name}/messages", "/workflows/{workflow_id}/runs/{run_id}", "/workflows/{workflow_id}/runs/{run_id}/history", diff --git a/src/update_outcome.rs b/src/update_outcome.rs new file mode 100644 index 00000000..2b2d9058 --- /dev/null +++ b/src/update_outcome.rs @@ -0,0 +1,368 @@ +//! Pending update outcomes and reading an accepted update's outcome later +//! by its update id (durable-workflow/.github#212, server#1131). +//! +//! A Server that ends a completion wait before the update handler finishes +//! answers HTTP 202 with no result. That is a pending update, not a handler +//! that returned null, so the client raises [`Error::UpdateStillPending`] +//! instead of returning `Null`. On a Server that advertises +//! `capabilities.update_outcome_reads`, [`Client::update_outcome`] reads the +//! same update again by its id without submitting it again. + +use std::{fmt, sync::atomic::Ordering, time::Duration}; + +use serde_json::Value; + +use crate::{ + decode_wire_avro_value, decode_wire_value, percent_encode_path_segment, AvroValue, Client, + Error, RequestProtocol, Result, WorkflowHandle, DEFAULT_CODEC, +}; + +/// The cluster capability that advertises `GET /workflows/{id}/updates/{update_id}`. +pub(crate) const UPDATE_OUTCOME_READS_CAPABILITY: &str = "update_outcome_reads"; + +/// Extra transport time allowed beyond a requested completion wait. +const WAIT_TRANSPORT_MARGIN: Duration = Duration::from_secs(30); + +/// How long an update outcome read waits before the Server answers. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[non_exhaustive] +pub enum UpdateWaitStage { + /// Answer at once with the update's current receipt (the Server default). + #[default] + Accepted, + /// Wait for the update's outcome, up to the wait timeout. + Completed, +} + +impl UpdateWaitStage { + /// The `wait_for` value sent to the Server. + pub fn as_str(self) -> &'static str { + match self { + Self::Accepted => "accepted", + Self::Completed => "completed", + } + } +} + +/// An update the Server accepted that had not completed when the wait ended. +/// +/// It has no result yet: this is never a handler that returned null. Read it +/// again by [`update_id`](Self::update_id) with [`Client::update_outcome`] on a +/// Server that advertises `update_outcome_reads` (native 3.6.0 and later). On +/// an older Server, call [`Client::update_workflow`] again with the same +/// [`request_id`](Self::request_id): the Server returns the retained update +/// instead of admitting a second one. +/// +/// `run_id` names the run that owns the update now, which can be a +/// continue-as-new successor. `receipt` is the Server's full answer. +#[derive(Clone, Debug, PartialEq)] +#[non_exhaustive] +pub struct UpdatePending { + pub workflow_id: String, + pub run_id: Option, + pub update_name: Option, + pub update_id: Option, + pub request_id: Option, + /// The Server's `update_status`, normally `accepted`. + pub update_status: Option, + /// Whether the Server's completion wait ran out. + pub wait_timed_out: bool, + pub receipt: Value, + from_outcome_read: bool, +} + +impl UpdatePending { + fn from_receipt(workflow_id: &str, receipt: Value, from_outcome_read: bool) -> Self { + let string = |name: &str| { + receipt + .get(name) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string) + }; + Self { + workflow_id: string("workflow_id").unwrap_or_else(|| workflow_id.to_string()), + run_id: string("run_id"), + update_name: string("update_name"), + update_id: string("update_id"), + request_id: string("request_id"), + update_status: string("update_status"), + wait_timed_out: receipt.get("wait_timed_out") == Some(&Value::Bool(true)), + receipt, + from_outcome_read, + } + } +} + +impl fmt::Display for UpdatePending { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let status = self.update_status.as_deref().unwrap_or("pending"); + let update = match (&self.update_name, &self.update_id) { + (Some(name), Some(id)) => format!("{name} ({id})"), + (Some(name), None) => name.clone(), + (None, Some(id)) => id.clone(), + (None, None) => "update".to_string(), + }; + write!( + f, + "update_still_pending: update {update} on workflow {} is still {status}; it has no result yet.", + self.workflow_id + )?; + match (&self.update_id, &self.request_id) { + (Some(id), _) if self.from_outcome_read => write!( + f, + " Call update_outcome() again with update id {id:?} to keep waiting. Reading never admits another update." + ), + (Some(id), Some(key)) => write!( + f, + " Read it again with update_outcome() and update id {id:?} on a Server that advertises update_outcome_reads, or call update_workflow() again with request_id {key:?}: the Server returns the retained update instead of admitting a second one." + ), + (Some(id), None) => write!( + f, + " Read it again with update_outcome() and update id {id:?} on a Server that advertises update_outcome_reads." + ), + (None, Some(key)) => write!( + f, + " Call update_workflow() again with request_id {key:?}: the Server returns the retained update instead of admitting a second one." + ), + (None, None) => Ok(()), + } + } +} + +impl std::error::Error for UpdatePending {} + +/// Whether a completion-wait receipt describes an update that has not finished. +/// +/// A receipt with a result envelope is finished (an encoded null is a real +/// null result). Otherwise a timed-out wait, or any status other than +/// `completed`, `failed` or `rejected`, is pending. +pub(crate) fn update_receipt_is_pending(receipt: &Value) -> bool { + if receipt.get("result_envelope").is_some_and(Value::is_object) { + return false; + } + receipt.get("wait_timed_out") == Some(&Value::Bool(true)) + || receipt + .get("update_status") + .and_then(Value::as_str) + .is_some_and(|status| !matches!(status, "completed" | "failed" | "rejected")) +} + +/// Raise [`Error::UpdateStillPending`] for a pending completion-wait receipt. +pub(crate) fn reject_pending_update( + workflow_id: &str, + receipt: Value, + from_outcome_read: bool, +) -> Result { + if update_receipt_is_pending(&receipt) { + return Err(Error::UpdateStillPending(UpdatePending::from_receipt( + workflow_id, + receipt, + from_outcome_read, + ))); + } + Ok(receipt) +} + +fn completed_update_avro_value(receipt: &Value) -> Result { + let envelope = receipt + .get("result_envelope") + .filter(|envelope| !envelope.is_null()) + .ok_or_else(|| { + Error::Codec( + "missing_payload_envelope: typed update result requires result_envelope" + .to_string(), + ) + })?; + decode_wire_avro_value(envelope, DEFAULT_CODEC) +} + +impl Client { + /// Read an accepted update's outcome later by its update id, without + /// submitting it again. + /// + /// This waits for the outcome, like [`Client::update_workflow`], for up to + /// `wait_timeout_seconds` (the Server's update wait default when `None`). + /// It returns the decoded result of a completed update, including `Null` + /// when the handler returned null. While the update is still running it + /// returns [`Error::UpdateStillPending`]: call it again with the same + /// update id to keep waiting. + /// + /// A failed update is [`Error::Http`] with status 422 and the failure + /// receipt, and a rejected update keeps its original rejection status. An + /// unknown or pruned update id is 404 `update_not_found` + /// ([`Error::is_update_not_found`]), which does not prove the update was + /// never admitted. + /// + /// Needs a Server that advertises `capabilities.update_outcome_reads` + /// (native 3.6.0 and later). Any other Server is refused with + /// [`Error::UpdateOutcomeReadsUnavailable`] before anything is read: the + /// SDK never resubmits an update as a substitute for reading it. + pub async fn update_outcome( + &self, + workflow_id: &str, + update_id: &str, + wait_timeout_seconds: Option, + ) -> Result { + let receipt = self + .completed_update_outcome(workflow_id, update_id, wait_timeout_seconds) + .await?; + match receipt + .get("result_envelope") + .filter(|envelope| !envelope.is_null()) + { + Some(envelope) => decode_wire_value(envelope, DEFAULT_CODEC), + None => Ok(receipt.get("result").cloned().unwrap_or(Value::Null)), + } + } + + /// [`Client::update_outcome`] that retains a bytes-capable Avro result. + pub async fn update_outcome_avro_value( + &self, + workflow_id: &str, + update_id: &str, + wait_timeout_seconds: Option, + ) -> Result { + let receipt = self + .completed_update_outcome(workflow_id, update_id, wait_timeout_seconds) + .await?; + completed_update_avro_value(&receipt) + } + + /// Read an accepted update's Server receipt by update id. It writes nothing. + /// + /// [`UpdateWaitStage::Accepted`] answers at once; a still-running update + /// comes back as a receipt with `update_status: "accepted"`, never as an + /// error. [`UpdateWaitStage::Completed`] waits up to + /// `wait_timeout_seconds` first. Failures, rejections, `update_not_found` + /// and the capability check behave as in [`Client::update_outcome`]. + pub async fn update_outcome_receipt( + &self, + workflow_id: &str, + update_id: &str, + wait_for: UpdateWaitStage, + wait_timeout_seconds: Option, + ) -> Result { + if update_id.trim().is_empty() { + return Err(Error::Codec( + "an update outcome read needs the update id".to_string(), + )); + } + self.require_update_outcome_reads().await?; + + let mut path = format!( + "/workflows/{}/updates/{}", + percent_encode_path_segment(workflow_id), + percent_encode_path_segment(update_id) + ); + let mut query = Vec::new(); + if wait_for == UpdateWaitStage::Completed { + query.push(format!("wait_for={}", wait_for.as_str())); + } + if let Some(seconds) = wait_timeout_seconds { + query.push(format!("wait_timeout_seconds={seconds}")); + } + if !query.is_empty() { + path.push('?'); + path.push_str(&query.join("&")); + } + let timeout = wait_timeout_seconds + .map(|seconds| Duration::from_secs(seconds).saturating_add(WAIT_TRANSPORT_MARGIN)) + .unwrap_or(Duration::ZERO) + .max(Duration::from_secs(60)); + self.request_json_with_timeout( + reqwest::Method::GET, + &path, + RequestProtocol::ControlPlane, + Option::<&Value>::None, + timeout, + ) + .await + } + + async fn completed_update_outcome( + &self, + workflow_id: &str, + update_id: &str, + wait_timeout_seconds: Option, + ) -> Result { + let receipt = self + .update_outcome_receipt( + workflow_id, + update_id, + UpdateWaitStage::Completed, + wait_timeout_seconds, + ) + .await?; + reject_pending_update(workflow_id, receipt, true) + } + + /// Refuse a Server that does not advertise `update_outcome_reads`. + /// + /// Discovery runs until a Server advertises the capability; after that the + /// client remembers it. + async fn require_update_outcome_reads(&self) -> Result<()> { + if self.update_outcome_reads_supported.load(Ordering::SeqCst) { + return Ok(()); + } + let discovery = self.cluster_info().await?; + if discovery["capabilities"][UPDATE_OUTCOME_READS_CAPABILITY] != Value::Bool(true) { + return Err(Error::UpdateOutcomeReadsUnavailable); + } + self.update_outcome_reads_supported + .store(true, Ordering::SeqCst); + Ok(()) + } +} + +impl WorkflowHandle { + /// Read one of this workflow's accepted updates by update id; see + /// [`Client::update_outcome`]. + pub async fn update_outcome( + &self, + update_id: &str, + wait_timeout_seconds: Option, + ) -> Result { + self.client + .update_outcome(&self.workflow_id, update_id, wait_timeout_seconds) + .await + } + + /// Read one of this workflow's accepted updates by update id with a + /// bytes-capable Avro result; see [`Client::update_outcome_avro_value`]. + pub async fn update_outcome_avro_value( + &self, + update_id: &str, + wait_timeout_seconds: Option, + ) -> Result { + self.client + .update_outcome_avro_value(&self.workflow_id, update_id, wait_timeout_seconds) + .await + } +} + +impl Client { + /// Decode a submitted update's completion-wait receipt. + pub(crate) fn submitted_update_value(workflow_id: &str, receipt: Value) -> Result { + let receipt = reject_pending_update(workflow_id, receipt, false)?; + if let Some(envelope) = receipt + .get("result_envelope") + .filter(|envelope| !envelope.is_null()) + { + return decode_wire_value(envelope, DEFAULT_CODEC); + } + Ok(receipt.get("result").cloned().unwrap_or(receipt)) + } + + /// Decode a submitted update's completion-wait receipt as a typed Avro value. + pub(crate) fn submitted_update_avro_value( + workflow_id: &str, + receipt: Value, + ) -> Result { + let receipt = reject_pending_update(workflow_id, receipt, false)?; + completed_update_avro_value(&receipt) + } +} + +#[cfg(test)] +mod tests; diff --git a/src/update_outcome/tests.rs b/src/update_outcome/tests.rs new file mode 100644 index 00000000..5f93d552 --- /dev/null +++ b/src/update_outcome/tests.rs @@ -0,0 +1,383 @@ +//! #212: pending updates are never a null result, and an accepted update's +//! outcome is read later by update id (server#1131), never by resubmitting. + +use std::time::Duration; + +use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use crate::{encode_typed_envelope, AvroValue, Client, Error, UpdateWaitStage, DEFAULT_CODEC}; + +/// One captured request: method, path with query, and body. +#[derive(Debug)] +struct Captured { + method: String, + path: String, + body: String, +} + +/// Answer each connection with the next scripted reply and capture the requests. +async fn server(replies: Vec<(u16, Value)>) -> (Client, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = Client::builder(format!("http://{}", listener.local_addr().unwrap())) + .timeout(Duration::from_secs(5)) + .build() + .unwrap(); + let requests = tokio::spawn(async move { + let mut captured = Vec::new(); + for (status, reply) in replies { + let (mut stream, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept()) + .await + .expect("the client sent another request") + .unwrap(); + let mut bytes = Vec::new(); + let (head, offset) = loop { + let mut buf = [0; 4096]; + let count = stream.read(&mut buf).await.unwrap(); + assert!(count > 0); + bytes.extend_from_slice(&buf[..count]); + if let Some(offset) = bytes.windows(4).position(|v| v == b"\r\n\r\n") { + break ( + String::from_utf8_lossy(&bytes[..offset]).to_string(), + offset + 4, + ); + } + }; + let length: usize = head + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse().unwrap()) + }) + .unwrap_or(0); + while bytes.len() < offset + length { + let mut buf = [0; 4096]; + let count = stream.read(&mut buf).await.unwrap(); + assert!(count > 0); + bytes.extend_from_slice(&buf[..count]); + } + let mut request_line = head.lines().next().unwrap().split_whitespace(); + captured.push(Captured { + method: request_line.next().unwrap().to_string(), + path: request_line.next().unwrap().to_string(), + body: String::from_utf8_lossy(&bytes[offset..offset + length]).to_string(), + }); + let body = reply.to_string(); + stream + .write_all( + format!( + "HTTP/1.1 {status} Reply\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + } + captured + }); + (client, requests) +} + +fn cluster() -> (u16, Value) { + ( + 200, + json!({"version": "3.6.0", "capabilities": {"update_outcome_reads": true}}), + ) +} + +fn envelope(value: AvroValue) -> Value { + encode_typed_envelope(&value, DEFAULT_CODEC).unwrap() +} + +fn pending_receipt() -> Value { + json!({ + "workflow_id": "order-1", "run_id": "run-9", "update_id": "upd-1", + "update_name": "rename", "request_id": "rename-1", "update_status": "accepted", + "result": null, "result_envelope": null, "wait_for": "completed", "wait_timed_out": true, + }) +} + +#[tokio::test] +async fn a_completed_read_returns_the_decoded_result_including_null() { + let (client, requests) = server(vec![ + cluster(), + ( + 200, + json!({"update_id": "upd-1", "update_status": "completed", "wait_timed_out": false, + "result": null, "result_envelope": envelope(AvroValue::String("renamed".into()))}), + ), + ( + 200, + json!({"update_id": "upd-2", "update_status": "completed", "wait_timed_out": false, + "result": null, "result_envelope": envelope(AvroValue::Null)}), + ), + ]) + .await; + + assert_eq!( + client + .update_outcome("order 1", "upd-1", Some(10)) + .await + .unwrap(), + json!("renamed") + ); + assert_eq!( + client + .update_outcome("order 1", "upd-2", None) + .await + .unwrap(), + Value::Null, + "a handler that returned null is a completed null result" + ); + + // Discovery happens once; each read is a GET on the update, never a POST. + let requests = requests.await.unwrap(); + assert_eq!(requests.len(), 3); + assert_eq!(requests[0].path, "/api/cluster/info"); + assert_eq!(requests[1].method, "GET"); + assert_eq!( + requests[1].path, + "/api/workflows/order%201/updates/upd-1?wait_for=completed&wait_timeout_seconds=10" + ); + assert!(requests[1].body.is_empty()); + assert_eq!(requests[2].method, "GET"); + assert_eq!( + requests[2].path, + "/api/workflows/order%201/updates/upd-2?wait_for=completed" + ); +} + +#[tokio::test] +async fn a_typed_completed_read_keeps_the_avro_value() { + let (client, _requests) = server(vec![ + cluster(), + ( + 200, + json!({"update_id": "upd-1", "update_status": "completed", + "result_envelope": envelope(AvroValue::Long(42))}), + ), + ]) + .await; + let handle = crate::WorkflowHandle { + client: client.clone(), + workflow_id: "order-1".to_string(), + run_id: None, + workflow_type: "orders".to_string(), + }; + + assert_eq!( + handle + .update_outcome_avro_value("upd-1", None) + .await + .unwrap(), + AvroValue::Long(42) + ); +} + +#[tokio::test] +async fn a_pending_read_is_a_typed_error_that_points_at_reading_again() { + let (client, _requests) = server(vec![cluster(), (202, pending_receipt())]).await; + + let error = client + .update_outcome("order-1", "upd-1", Some(1)) + .await + .expect_err("a pending update must not return a value"); + let Error::UpdateStillPending(pending) = &error else { + panic!("expected UpdateStillPending, got {error:?}"); + }; + assert_eq!(pending.workflow_id, "order-1"); + assert_eq!(pending.run_id.as_deref(), Some("run-9")); + assert_eq!(pending.update_id.as_deref(), Some("upd-1")); + assert_eq!(pending.update_name.as_deref(), Some("rename")); + assert_eq!(pending.request_id.as_deref(), Some("rename-1")); + assert_eq!(pending.update_status.as_deref(), Some("accepted")); + assert!(pending.wait_timed_out); + assert_eq!(pending.receipt, pending_receipt()); + let message = error.to_string(); + assert!( + message.contains("update_outcome() again with update id \"upd-1\""), + "{message}" + ); + assert!(message.contains("never admits another update"), "{message}"); +} + +#[tokio::test] +async fn reading_again_after_pending_returns_the_result() { + let (client, requests) = server(vec![ + cluster(), + (202, pending_receipt()), + ( + 200, + json!({"update_id": "upd-1", "update_status": "completed", + "result_envelope": envelope(AvroValue::Boolean(true))}), + ), + ]) + .await; + + let Err(Error::UpdateStillPending(pending)) = + client.update_outcome("order-1", "upd-1", None).await + else { + panic!("the first read was still pending"); + }; + let update_id = pending.update_id.expect("pending update id"); + assert_eq!( + client + .update_outcome("order-1", &update_id, None) + .await + .unwrap(), + json!(true) + ); + let requests = requests.await.unwrap(); + assert!( + requests.iter().all(|request| request.method == "GET"), + "reading never submits the update again: {requests:?}" + ); +} + +#[tokio::test] +async fn an_accepted_receipt_read_returns_the_receipt_without_waiting() { + let receipt = json!({"update_id": "upd-1", "update_status": "accepted", + "result_envelope": null, "wait_timed_out": false}); + let (client, requests) = server(vec![cluster(), (202, receipt.clone())]).await; + + assert_eq!( + client + .update_outcome_receipt("order-1", "upd-1", UpdateWaitStage::Accepted, None) + .await + .unwrap(), + receipt + ); + let requests = requests.await.unwrap(); + assert_eq!(requests[1].path, "/api/workflows/order-1/updates/upd-1"); +} + +#[tokio::test] +async fn a_failed_update_surfaces_its_failure_receipt() { + let failure = json!({"update_id": "upd-1", "update_status": "failed", "reason": "update_failed", + "failure_id": "fail-1", "failure_message": "boom"}); + let (client, _requests) = server(vec![cluster(), (422, failure.clone())]).await; + + let error = client + .update_outcome("order-1", "upd-1", None) + .await + .expect_err("a failed update has no result"); + let Error::Http { status, body } = &error else { + panic!("expected the failure receipt, got {error:?}"); + }; + assert_eq!(status.as_u16(), 422); + assert_eq!(serde_json::from_str::(body).unwrap(), failure); + assert!(!error.is_update_not_found()); +} + +#[tokio::test] +async fn an_unknown_update_is_update_not_found() { + let (client, _requests) = server(vec![ + cluster(), + ( + 404, + json!({"reason": "update_not_found", "workflow_id": "order-1", "update_id": "upd-x", + "message": "No retained update has this id."}), + ), + ]) + .await; + + let error = client + .update_outcome("order-1", "upd-x", None) + .await + .expect_err("an unknown update has no outcome"); + assert!(error.is_update_not_found(), "{error:?}"); +} + +#[tokio::test] +async fn a_server_without_the_capability_is_refused_before_any_read() { + let (client, requests) = server(vec![( + 200, + json!({"version": "3.5.0", "capabilities": {"command_request_keys": false}}), + )]) + .await; + + let error = client + .update_outcome("order-1", "upd-1", None) + .await + .expect_err("an older Server must be refused"); + assert!( + matches!(error, Error::UpdateOutcomeReadsUnavailable), + "{error:?}" + ); + assert!(error.to_string().contains("update_outcome_reads")); + let requests = requests.await.unwrap(); + assert_eq!( + requests.len(), + 1, + "only discovery ran: nothing was read or resubmitted" + ); + assert_eq!(requests[0].path, "/api/cluster/info"); +} + +#[tokio::test] +async fn an_empty_update_id_is_refused_without_a_request() { + let (client, requests) = server(vec![]).await; + let error = client + .update_outcome_receipt("order-1", " ", UpdateWaitStage::Completed, None) + .await + .expect_err("an update id is required"); + assert!(matches!(error, Error::Codec(_)), "{error:?}"); + assert!(requests.await.unwrap().is_empty()); +} + +#[tokio::test] +async fn a_submitted_update_that_times_out_is_pending_not_null() { + // The first inline payload discovers the request policy from cluster info. + let (client, requests) = server(vec![ + (200, json!({"version": "3.5.0", "capabilities": {}})), + (202, pending_receipt()), + (202, pending_receipt()), + ]) + .await; + + let error = client + .update_workflow("order-1", "rename", json!(["new"]), Some("rename-1")) + .await + .expect_err("a timed-out update is not a null result"); + let Error::UpdateStillPending(pending) = &error else { + panic!("expected UpdateStillPending, got {error:?}"); + }; + assert_eq!(pending.update_id.as_deref(), Some("upd-1")); + assert_eq!(pending.request_id.as_deref(), Some("rename-1")); + let message = error.to_string(); + assert!(message.contains("update id \"upd-1\""), "{message}"); + assert!(message.contains("request_id \"rename-1\""), "{message}"); + + let typed = client + .update_workflow_avro_value("order-1", "rename", json!(["new"]), Some("rename-1")) + .await + .expect_err("a timed-out typed update is pending, not a codec error"); + assert!(matches!(typed, Error::UpdateStillPending(_)), "{typed:?}"); + let requests = requests.await.unwrap(); + assert_eq!(requests[0].path, "/api/cluster/info"); + assert!(requests[1..].iter().all(|request| request.method == "POST" + && request.path == "/api/workflows/order-1/update/rename")); +} + +#[tokio::test] +async fn a_submitted_update_that_returned_null_is_a_null_result() { + let (client, _requests) = server(vec![ + (200, json!({"version": "3.5.0", "capabilities": {}})), + ( + 200, + json!({"workflow_id": "order-1", "update_id": "upd-1", "update_status": "completed", + "wait_timed_out": false, "result": null, "result_envelope": envelope(AvroValue::Null)}), + ), + ]) + .await; + + assert_eq!( + client + .update_workflow("order-1", "rename", json!(["new"]), Some("rename-1")) + .await + .unwrap(), + Value::Null + ); +} From c4c96e14494737ce8de0a5f747481ff5b3cf2097 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 05:15:23 +0000 Subject: [PATCH 2/2] Refuse dot-segment ids and unrecognized answers in update outcome reads (#212) Security review of #243: an update id of "." or ".." collapsed the read onto the workflow describe route, and any 200 body without update_status was treated as finished, so update_outcome could return the workflow's result or Null for an update it never read. Workflow and update ids now go through RawRequest's segment check and an empty, "." or ".." id is refused before any request. A successful answer must be a receipt for the requested update (matching update_id, an update_status, and the workflow id when echoed), and update_outcome only returns a value from a completed receipt; anything else is Error::UnrecognizedUpdateOutcome. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 ++- README.md | 4 +- src/lib.rs | 12 +++++ src/update_outcome.rs | 67 +++++++++++++++++++------ src/update_outcome/tests.rs | 98 +++++++++++++++++++++++++++++++++++++ 5 files changed, 170 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 42cf69e0..a92976ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,12 @@ `Error::UpdateOutcomeReadsUnavailable` otherwise, before reading anything. A still-running update returns `Error::UpdateStillPending` with a hint to read again by update id. A failed update is `Error::Http` 422, and - `Error::is_update_not_found()` detects 404 `update_not_found`. + `Error::is_update_not_found()` detects 404 `update_not_found`. An empty, `.` + or `..` workflow id or update id is refused before any request, so a read + can't collapse onto another route. A successful answer that isn't a receipt + for the requested update (matching `update_id` and an `update_status`), or a + non-completed receipt where an outcome was expected, returns + `Error::UnrecognizedUpdateOutcome` instead of a value. - A `request_id_conflict` refusal from native Server 3.9.0 (server#934, server#1191) now names the key that was sent. Native doesn't echo the key, so diff --git a/README.md b/README.md index 2ba4051b..e5bb5d7f 100644 --- a/README.md +++ b/README.md @@ -156,7 +156,9 @@ while the update is still running. A failed update is `Error::Http` with status 422, and an unknown or pruned update id is 404 `update_not_found` (`Error::is_update_not_found()`), which does not prove the update was never admitted. `update_outcome_receipt` returns the raw receipt and can answer at -once with `UpdateWaitStage::Accepted`. +once with `UpdateWaitStage::Accepted`. Workflow and update ids must each be one +path segment (not empty, `.` or `..`), and an answer that is not a receipt for +the requested update is `Error::UnrecognizedUpdateOutcome`, never a result. On any other Server, `update_outcome` returns `Error::UpdateOutcomeReadsUnavailable` before reading anything; the SDK never diff --git a/src/lib.rs b/src/lib.rs index a77ae82f..0676c001 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -420,6 +420,18 @@ pub enum Error { "update_outcome_reads_unavailable: the connected Server does not advertise update_outcome_reads, so an update outcome cannot be read by update id; call update_workflow() again with the same request_id instead" )] UpdateOutcomeReadsUnavailable, + /// A successful update outcome read did not answer with a receipt for the + /// requested update: it lacked a matching `update_id` and `update_status`, + /// or claimed a finished outcome without being completed. No outcome is + /// returned for it. + #[error( + "update_outcome_unrecognized: the Server's answer is not an outcome receipt for update {update_id:?} on workflow {workflow_id:?}; no outcome was read" + )] + UnrecognizedUpdateOutcome { + workflow_id: String, + update_id: String, + receipt: Value, + }, #[error("cooperative cancellation is unavailable: {0}")] CooperativeCancellationUnavailable(String), #[error("cancellation_scope_execution_not_supported: Rust worker cannot replay scoped cancellation history")] diff --git a/src/update_outcome.rs b/src/update_outcome.rs index 2b2d9058..295bbc52 100644 --- a/src/update_outcome.rs +++ b/src/update_outcome.rs @@ -13,8 +13,8 @@ use std::{fmt, sync::atomic::Ordering, time::Duration}; use serde_json::Value; use crate::{ - decode_wire_avro_value, decode_wire_value, percent_encode_path_segment, AvroValue, Client, - Error, RequestProtocol, Result, WorkflowHandle, DEFAULT_CODEC, + decode_wire_avro_value, decode_wire_value, AvroValue, Client, Error, RawRequest, + RequestProtocol, Result, WorkflowHandle, DEFAULT_CODEC, }; /// The cluster capability that advertises `GET /workflows/{id}/updates/{update_id}`. @@ -164,6 +164,24 @@ pub(crate) fn reject_pending_update( Ok(receipt) } +/// Whether a successful answer is the Server's receipt for the update that was read. +fn is_receipt_for(receipt: &Value, workflow_id: &str, update_id: &str) -> bool { + receipt.get("update_id").and_then(Value::as_str) == Some(update_id) + && receipt.get("update_status").is_some_and(Value::is_string) + && receipt + .get("workflow_id") + .and_then(Value::as_str) + .is_none_or(|echoed| echoed == workflow_id) +} + +fn unrecognized(workflow_id: &str, update_id: &str, receipt: Value) -> Error { + Error::UnrecognizedUpdateOutcome { + workflow_id: workflow_id.to_string(), + update_id: update_id.to_string(), + receipt, + } +} + fn completed_update_avro_value(receipt: &Value) -> Result { let envelope = receipt .get("result_envelope") @@ -248,13 +266,17 @@ impl Client { "an update outcome read needs the update id".to_string(), )); } + // Each id must stay one path segment: "." or ".." would collapse the + // read onto another route, such as the workflow description. + let mut path = RawRequest::get(["workflows", workflow_id, "updates", update_id]) + .path() + .map_err(|_| { + Error::Codec( + "an update outcome read needs a workflow id and update id that are each one path segment: not empty, \".\" or \"..\"".to_string(), + ) + })?; self.require_update_outcome_reads().await?; - let mut path = format!( - "/workflows/{}/updates/{}", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(update_id) - ); let mut query = Vec::new(); if wait_for == UpdateWaitStage::Completed { query.push(format!("wait_for={}", wait_for.as_str())); @@ -270,14 +292,19 @@ impl Client { .map(|seconds| Duration::from_secs(seconds).saturating_add(WAIT_TRANSPORT_MARGIN)) .unwrap_or(Duration::ZERO) .max(Duration::from_secs(60)); - self.request_json_with_timeout( - reqwest::Method::GET, - &path, - RequestProtocol::ControlPlane, - Option::<&Value>::None, - timeout, - ) - .await + let receipt: Value = self + .request_json_with_timeout( + reqwest::Method::GET, + &path, + RequestProtocol::ControlPlane, + Option::<&Value>::None, + timeout, + ) + .await?; + if !is_receipt_for(&receipt, workflow_id, update_id) { + return Err(unrecognized(workflow_id, update_id, receipt)); + } + Ok(receipt) } async fn completed_update_outcome( @@ -294,7 +321,15 @@ impl Client { wait_timeout_seconds, ) .await?; - reject_pending_update(workflow_id, receipt, true) + let receipt = reject_pending_update(workflow_id, receipt, true)?; + // Only a completed receipt carries an outcome; never read one into + // any other successful answer. + let completed = receipt.get("result_envelope").is_some_and(Value::is_object) + || receipt.get("update_status").and_then(Value::as_str) == Some("completed"); + if !completed { + return Err(unrecognized(workflow_id, update_id, receipt)); + } + Ok(receipt) } /// Refuse a Server that does not advertise `update_outcome_reads`. diff --git a/src/update_outcome/tests.rs b/src/update_outcome/tests.rs index 5f93d552..6dd1b1a2 100644 --- a/src/update_outcome/tests.rs +++ b/src/update_outcome/tests.rs @@ -381,3 +381,101 @@ async fn a_submitted_update_that_returned_null_is_a_null_result() { Value::Null ); } + +#[tokio::test] +async fn dot_segment_and_empty_ids_are_refused_before_any_request() { + let (client, requests) = server(vec![]).await; + for (workflow_id, update_id) in [ + ("order-1", "."), + ("order-1", ".."), + (".", "upd-1"), + ("..", "upd-1"), + ("", "upd-1"), + ] { + let error = client + .update_outcome(workflow_id, update_id, None) + .await + .expect_err("a dot segment would read another route"); + assert!( + matches!(error, Error::Codec(_)), + "{workflow_id:?}/{update_id:?}: {error:?}" + ); + let error = client + .update_outcome_receipt(workflow_id, update_id, UpdateWaitStage::Accepted, None) + .await + .expect_err("a dot segment would read another route"); + assert!(matches!(error, Error::Codec(_)), "{error:?}"); + } + assert!( + requests.await.unwrap().is_empty(), + "nothing was sent, not even discovery" + ); +} + +#[tokio::test] +async fn a_success_that_is_not_a_receipt_for_the_update_is_never_an_outcome() { + let describe = json!({"workflow_id": "order-1", "run_id": "run-9", "status": "completed", + "result": "the workflow's own result"}); + let other_update = json!({"workflow_id": "order-1", "update_id": "upd-2", + "update_status": "completed", "result_envelope": envelope(AvroValue::Long(7))}); + let other_workflow = json!({"workflow_id": "order-2", "update_id": "upd-1", + "update_status": "completed", "result_envelope": envelope(AvroValue::Long(7))}); + let (client, _requests) = server(vec![ + cluster(), + (200, describe.clone()), + (200, json!({})), + (200, other_update), + (200, other_workflow), + (200, describe.clone()), + ]) + .await; + + for _ in 0..4 { + let error = client + .update_outcome("order-1", "upd-1", None) + .await + .expect_err("only a receipt for upd-1 carries its outcome"); + let Error::UnrecognizedUpdateOutcome { + workflow_id, + update_id, + .. + } = &error + else { + panic!("expected UnrecognizedUpdateOutcome, got {error:?}"); + }; + assert_eq!( + (workflow_id.as_str(), update_id.as_str()), + ("order-1", "upd-1") + ); + } + let error = client + .update_outcome_receipt("order-1", "upd-1", UpdateWaitStage::Accepted, None) + .await + .expect_err("a describe body is not a receipt"); + let Error::UnrecognizedUpdateOutcome { receipt, .. } = error else { + panic!("expected UnrecognizedUpdateOutcome"); + }; + assert_eq!(receipt, describe); +} + +#[tokio::test] +async fn a_successful_receipt_that_is_not_completed_is_never_an_outcome() { + let (client, _requests) = server(vec![ + cluster(), + ( + 200, + json!({"workflow_id": "order-1", "update_id": "upd-1", "update_status": "failed", + "wait_timed_out": false, "result": null, "result_envelope": null}), + ), + ]) + .await; + + let error = client + .update_outcome("order-1", "upd-1", None) + .await + .expect_err("a failed status is not a null result"); + assert!( + matches!(error, Error::UnrecognizedUpdateOutcome { .. }), + "{error:?}" + ); +}