diff --git a/CHANGELOG.md b/CHANGELOG.md index 77c2eb6b..08f0c6e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,30 @@ ## 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`. 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. + - `describe_workflow` and `update_workflow` (and `update_workflow_avro_value`) percent-encode the workflow id, and the update name, as one path segment. Before, they were put into the path raw, so an id containing `/`, `?` or `#` diff --git a/README.md b/README.md index d544c712..e5bb5d7f 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,50 @@ 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`. 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 +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 5d222c18..ce393a67 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}; @@ -425,6 +427,28 @@ 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, + /// 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")] @@ -515,6 +539,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(), @@ -3113,6 +3155,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")] @@ -3706,6 +3750,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, @@ -3716,13 +3766,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. @@ -3736,16 +3780,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( @@ -6153,6 +6188,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..295bbc52 --- /dev/null +++ b/src/update_outcome.rs @@ -0,0 +1,403 @@ +//! 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, AvroValue, Client, Error, RawRequest, + 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) +} + +/// 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") + .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(), + )); + } + // 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 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)); + 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( + &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?; + 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`. + /// + /// 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..6dd1b1a2 --- /dev/null +++ b/src/update_outcome/tests.rs @@ -0,0 +1,481 @@ +//! #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 + ); +} + +#[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:?}" + ); +}