Return pending updates as UpdateStillPending and read outcomes later by id

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 `#`

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

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<Value> {
         match self {
             Self::Http { body, .. } => serde_json::from_str(body).ok(),
@@ -3113,6 +3155,8 @@ pub struct Client {
     activity_lease_renewal_supported: Arc<AtomicBool>,
     /// Server #1017: the Server sends sticky hits only the history after the cached sequence.
     sticky_history_suffix_supported: Arc<AtomicBool>,
+    /// Server #1131: the Server advertised `update_outcome_reads`.
+    update_outcome_reads_supported: Arc<AtomicBool>,
     activity_lease_discovery: Arc<activity_lease::Discovery>,
     activity_lease_authority: Option<Arc<activity_lease::authority::Authority>>,
     #[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<T: Serialize>(
         &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<T: Serialize>(
@@ -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")]

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",