Merge branch 'rust-325-update-validation-terminal' into rust-325-worker-poll-fence-native

Author: rmcdaniel

rust/src/runtime/mod.rs

@@ -80,6 +80,9 @@ mod operator_metrics;
 mod payload_cleanup;
 mod payload_discovery;
 mod payload_limits;
+#[cfg(debug_assertions)]
+#[doc(hidden)]
+pub mod poll_pauses;
 mod poll_readiness;
 pub mod postgres;
 mod previews;

rust/src/runtime/poll_pauses.rs

@@ -0,0 +1,62 @@
+//! A debug-build test seam for workflow polls that carry a request ID. A test
+//! registers a stop for one request ID at one point; the next poll with that
+//! ID to reach the point reports its arrival and waits there until the test
+//! lets it go. Regressions use it to commit a duplicate's work inside a window
+//! between a poll's steps, in a fixed order and without sleeping. Release
+//! builds compile neither this module nor its call sites.
+use std::sync::{Mutex, PoisonError};
+use tokio::sync::oneshot;
+
+/// Where a workflow poll with a request ID can be stopped.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PollPoint {
+    /// A held single-kind poll whose probe found nothing and whose queue
+    /// had not changed, before it waits for work.
+    Waiting,
+    /// After the poll's last probe and its decision to answer without a
+    /// task, before it records that answer under the request ID.
+    Terminal,
+}
+
+/// A registered stop. `reached` resolves when the poll arrives; sending on
+/// `release`, or dropping it, lets the poll go on.
+pub struct PollPause {
+    pub reached: oneshot::Receiver<()>,
+    pub release: oneshot::Sender<()>,
+}
+
+struct Stop {
+    request_id: String,
+    point: PollPoint,
+    arrived: oneshot::Sender<()>,
+    resume: oneshot::Receiver<()>,
+}
+
+static STOPS: Mutex<Vec<Stop>> = Mutex::new(Vec::new());
+
+/// Stop the next workflow poll with `request_id` at `point`, once. The stops
+/// are process-wide, so each test uses request IDs of its own.
+pub fn pause_workflow_poll(request_id: &str, point: PollPoint) -> PollPause {
+    let (arrived, reached) = oneshot::channel();
+    let (release, resume) = oneshot::channel();
+    STOPS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+        .push(Stop { request_id: request_id.to_owned(), point, arrived, resume });
+    PollPause { reached, release }
+}
+
+/// A poll at `point`: wait there if a test registered a stop for its ID.
+pub(super) async fn stop_at(request_id: &str, point: PollPoint) {
+    let stop = {
+        let mut stops = STOPS.lock().unwrap_or_else(PoisonError::into_inner);
+        stops
+            .iter()
+            .position(|stop| stop.request_id == request_id && stop.point == point)
+            .map(|index| stops.swap_remove(index))
+    };
+    if let Some(stop) = stop {
+        let _ = stop.arrived.send(());
+        let _ = stop.resume.await;
+    }
+}

rust/src/runtime/store.rs

@@ -1474,12 +1474,8 @@ where
                     return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch"));
                 }
                 let request = self.workflow_poll_request(worker_id, queue, id, default_wait, &["workflow"]).await?;
-                if let Some(mut replay) = self.replay_workflow_poll(&request, worker_id, queue).await? {
-                    if replay["task"].is_object() {
-                        super::history_pages::compress(&mut replay["task"], encoding)?;
-                    }
-                    replay["server_capabilities"] = capabilities();
-                    return Ok(replay);
+                if let Some(replay) = self.replay_workflow_poll(&request, worker_id, queue).await? {
+                    return replayed_poll(replay, encoding);
                 }
                 Some(request)
             }
@@ -1531,29 +1527,41 @@ where
                 );
             }
             if tokio::time::Instant::now() >= deadline {
-                if let Some(request) = &poll_request {
-                    self.remember_empty_workflow_poll(request, "empty").await?;
-                }
-                return Ok(
-                    json!({"task": null, "poll_status": "empty", "server_capabilities": capabilities()}),
-                );
+                return self.no_task_answer(&body, poll_request.as_ref(), "empty", encoding).await;
             }
             if self
                 .poll_queue_changed(&body, kind, &mut cross_kind)
                 .await?
                 && tokio::time::Instant::now() < deadline
             {
-                if let Some(request) = &poll_request {
-                    self.remember_empty_workflow_poll(request, "task_queue_changed").await?;
-                }
-                return Ok(
-                    json!({"task":null,"poll_status":"task_queue_changed","server_capabilities":capabilities()}),
-                );
+                return self.no_task_answer(&body, poll_request.as_ref(), "task_queue_changed", encoding).await;
+            }
+            #[cfg(debug_assertions)]
+            if let Some(request) = &poll_request {
+                super::poll_pauses::stop_at(&request.id, super::poll_pauses::PollPoint::Waiting).await;
             }
             pace.wait(wake, deadline).await;
         }
     }
 
+    /// A poll's answer when it leased nothing. Under a request ID, a
+    /// duplicate's lease committed after this poll's last probe replays
+    /// instead, as a retry with that ID would see it.
+    async fn no_task_answer(
+        &self,
+        body: &Value,
+        poll_request: Option<&super::workflow_poll_requests::WorkflowPollRequest>,
+        poll_status: &str,
+        encoding: Option<&'static str>,
+    ) -> Result<Value> {
+        if let Some(request) = poll_request
+            && let Some(lease) = self.remember_empty_workflow_poll(request, text(body, "worker_id")?, text(body, "task_queue")?, poll_status).await?
+        {
+            return replayed_poll(lease, encoding);
+        }
+        Ok(json!({"task": null, "poll_status": poll_status, "server_capabilities": capabilities()}))
+    }
+
     /// Whether a held workflow or activity poll should claim again.
     async fn task_poll_ready(
         &self,
@@ -1659,8 +1667,7 @@ where
         // replays a validation answer itself before claiming.
         if let Some(request) = poll_request {
             Self::bind_workflow_poll(tx, request).await?;
-            if let Some(answer) = self.deliverable_workflow_poll(tx, request, worker_id, queue).await?
-                && answer["task"].is_object()
+            if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, queue).await?
                 && answer["task"]["task_kind"] != "update_validation"
             {
                 return Ok(Some(answer["task"].clone()));
@@ -4176,6 +4183,16 @@ pub(super) fn reject_fields(body: &Value, allowed: &[&str]) -> Result<()> {
     Ok(())
 }
 
+/// A replayed answer as a workflow poll delivers it: a lease takes this
+/// request's history encoding.
+pub(super) fn replayed_poll(mut replay: Value, encoding: Option<&'static str>) -> Result<Value> {
+    if replay["task"].is_object() {
+        super::history_pages::compress(&mut replay["task"], encoding)?;
+    }
+    replay["server_capabilities"] = capabilities();
+    Ok(replay)
+}
+
 pub(crate) fn capabilities() -> Value {
     let mut value = json!({"supported_workflow_task_commands": ["schedule_activity", "start_timer", "start_child_workflow", "open_condition_wait", "open_signal_wait", "record_version_marker", "record_side_effect", "complete_workflow", "fail_workflow", "continue_as_new", "complete_update", "fail_update"],
         "worker_deregistration_fencing":{"schema":"durable-workflow.v2.worker-deregistration.v1","supported":true,"receipt_retention_seconds":600,"endpoint":"/worker/registrations/{workerId}/deregister"},

rust/src/runtime/update_validation.rs

@@ -938,15 +938,11 @@ where
 
     /// A replayed answer as the route delivers it: a validation lease is
     /// rebuilt from its task, any other answer is returned as cached.
-    fn replayed(mut answer: Value, encoding: Option<&'static str>) -> Result<Leased> {
+    fn replayed(answer: Value, encoding: Option<&'static str>) -> Result<Leased> {
         if answer["task"]["task_kind"] == "update_validation" {
             return Ok(Leased::Validation(text(&answer["task"], "update_validation_task_id")?.to_owned()));
         }
-        if answer["task"].is_object() {
-            super::history_pages::compress(&mut answer["task"], encoding)?;
-        }
-        answer["server_capabilities"] = super::store::capabilities();
-        Ok(Leased::Workflow(answer))
+        super::store::replayed_poll(answer, encoding).map(Leased::Workflow)
     }
 
     /// `WorkflowTaskPoller` with validation tasks requested. A request ID
@@ -955,7 +951,9 @@ where
     /// transaction: replay a duplicate's answer, lock the queue's cursor,
     /// claim the kind it names first and then the other, turn the cursor
     /// after a lease, and record the answer under the request ID. When
-    /// neither kind has work, the workflow poll's answer stands.
+    /// neither kind has work, the workflow poll's answer stands, unless a
+    /// duplicate leased under the request ID after the last probe: that
+    /// lease replays.
     pub(super) async fn poll_task_kinds(
         &self,
         body: &Value,
@@ -1013,8 +1011,10 @@ where
                 None => {}
             }
             if tokio::time::Instant::now() >= deadline {
-                if let Some(request) = &request {
-                    self.remember_empty_workflow_poll(request, "empty").await?;
+                if let Some(request) = &request
+                    && let Some(lease) = self.remember_empty_workflow_poll(request, &worker_id, &task_queue, "empty").await?
+                {
+                    return Self::replayed(lease, encoding);
                 }
                 return Ok(Leased::Workflow(json!({"task":null,"poll_status":"empty","server_capabilities":super::store::capabilities()})));
             }
@@ -1044,9 +1044,7 @@ where
         // binding row, locked first, makes it wait for that one's answer.
         if let Some(request) = request {
             Self::bind_workflow_poll(&mut tx, request).await?;
-            if let Some(answer) = self.deliverable_workflow_poll(&mut tx, request, worker_id, task_queue).await?
-                && answer["task"].is_object()
-            {
+            if let Some(answer) = self.deliverable_workflow_lease(&mut tx, request, worker_id, task_queue).await? {
                 tx.commit().await?;
                 return Ok(Some(Probe::Replayed(answer)));
             }

rust/src/runtime/workflow_poll_requests.rs

@@ -8,6 +8,12 @@
 //! validation lease while it stays leased at the same attempt. Once a lease is
 //! no longer deliverable, the ID polls afresh. PHP never refuses a retry as
 //! expired.
+//!
+//! PHP elects one pending leader per ID, and duplicates only read its answer.
+//! Native duplicates each probe, so every write of an answer first replays a
+//! deliverable lease a duplicate committed: a claim replays it instead of
+//! leasing again, and a poll that found nothing returns it instead of
+//! recording `empty` or `task_queue_changed` over it.
 use super::transaction::Transaction;
 use super::{Result, RuntimeError, store::now};
 use axum::http::StatusCode;
@@ -193,6 +199,20 @@ store_methods! {
         Ok(None)
     }
 
+    /// A deliverable lease a duplicate left under this request ID, which a
+    /// claim or a no-task answer must replay rather than overwrite. Call it
+    /// after `bind_workflow_poll`, whose lock orders it after any duplicate's
+    /// committed answer.
+    pub(super) async fn deliverable_workflow_lease(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        request: &WorkflowPollRequest,
+        worker_id: &str,
+        queue: &str,
+    ) -> Result<Option<Value>> {
+        Ok(self.deliverable_workflow_poll(tx, request, worker_id, queue).await?.filter(|answer| answer["task"].is_object()))
+    }
+
     /// `cachedTaskStillDeliverable` for a workflow task: leased on this
     /// queue to this worker at the delivered attempt, with a live lease,
     /// which the replay then shows.
@@ -260,12 +280,27 @@ store_methods! {
         })
     }
 
-    /// Remember an answer that leased nothing, in its own transaction.
-    pub(super) async fn remember_empty_workflow_poll(&self, request: &WorkflowPollRequest, poll_status: &str) -> Result<()> {
+    /// Remember an answer that leased nothing (`empty` or
+    /// `task_queue_changed`), in its own transaction. A duplicate may have
+    /// leased under this ID after this poll's last probe; its deliverable
+    /// lease is returned for the caller to replay, and stays recorded.
+    pub(super) async fn remember_empty_workflow_poll(
+        &self,
+        request: &WorkflowPollRequest,
+        worker_id: &str,
+        queue: &str,
+        poll_status: &str,
+    ) -> Result<Option<Value>> {
+        #[cfg(debug_assertions)]
+        super::poll_pauses::stop_at(&request.id, super::poll_pauses::PollPoint::Terminal).await;
         let mut tx = self.begin().await?;
         Self::bind_workflow_poll(&mut tx, request).await?;
-        Self::remember_workflow_poll(&mut tx, request, &json!({"task": null, "poll_status": poll_status})).await?;
-        tx.commit().await
+        let lease = self.deliverable_workflow_lease(&mut tx, request, worker_id, queue).await?;
+        if lease.is_none() {
+            Self::remember_workflow_poll(&mut tx, request, &json!({"task": null, "poll_status": poll_status})).await?;
+        }
+        tx.commit().await?;
+        Ok(lease)
     }
 
     async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str) -> Result<Option<Value>> {

rust/tests/execution/update_validation.rs

@@ -505,3 +505,64 @@ async fn concurrent_probes_alternate_kinds_without_losing_a_turn() {
         }
     }
 }
+
+// A duplicate can lease either kind after a multiplexed poll's last probe and
+// before that poll records `empty`. The held poll then returns the
+// duplicate's lease, and a retry whose original response was lost recovers
+// it. Debug builds stop the held poll between the two steps.
+#[cfg(debug_assertions)]
+mod terminal_races {
+    use super::*;
+    use durable_workflow_server::runtime::poll_pauses::{PollPoint, pause_workflow_poll};
+
+    async fn multiplexed(app: &Router, worker: &str, queue: &str, request_id: &str, timeout: u64) -> (StatusCode, Value) {
+        request(app, "POST", "/api/worker/workflow-tasks/poll", json!({"worker_id":worker,"task_queue":queue,
+            "timeout_seconds":timeout,"poll_request_id":request_id,"task_kinds":["workflow","update_validation"]})).await
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn a_lease_committed_before_the_empty_answer_replays_for_either_kind() {
+        for kind in ["update_validation", "workflow"] {
+            let database = TestDatabase::new().await;
+            let runtime = database.open().await.unwrap();
+            let app = router(runtime.clone());
+            let (worker, queue, run) = ("late", format!("q-late-{kind}"), format!("late-run-{kind}"));
+            validating_worker(&app, worker, &queue, Some(json!(["approve_me"])), true).await;
+            waiting_run(&app, &run, &queue, worker).await;
+            let id = format!("terminal-race-multiplexed-{kind}");
+            let mut terminal = pause_workflow_poll(&id, PollPoint::Terminal);
+
+            // The held poll finds neither kind and stops before answering.
+            let held = {
+                let (app, queue, id) = (app.clone(), queue.clone(), id.clone());
+                tokio::spawn(async move { multiplexed(&app, worker, &queue, &id, 0).await })
+            };
+            (&mut terminal.reached).await.unwrap();
+            // Work of one kind arrives, and a duplicate waits for it and
+            // leases it; its response is lost.
+            let pending = if kind == "update_validation" {
+                Some(send_update(&app, &run, "approve_me", update_body(7, Some("late-update"))))
+            } else {
+                start_on(&app, &format!("late-ready-{kind}"), &queue).await;
+                None
+            };
+            let (status, leased) = multiplexed(&app, worker, &queue, &id, 10).await;
+            assert_eq!((status, leased["poll_status"].clone(), delivery(&leased).0), (StatusCode::OK, json!("leased"), json!(kind)), "{leased}");
+            terminal.release.send(()).unwrap();
+
+            let (status, answered) = held.await.unwrap();
+            assert_eq!(status, StatusCode::OK, "{answered}");
+            assert_eq!((answered["poll_status"].clone(), delivery(&answered)), (json!("leased"), delivery(&leased)), "{kind}: {answered}");
+            let (status, retried) = multiplexed(&app, worker, &queue, &id, 0).await;
+            assert_eq!((status, delivery(&retried)), (StatusCode::OK, delivery(&leased)), "{kind}: {retried}");
+            if let Some(pending) = pending {
+                let task = delivery(&leased).1;
+                let approved = complete(&app, task.as_str().unwrap(), "approve", json!({"lease_owner":worker,"update_validation_attempt":1})).await;
+                assert_eq!(approved.0, StatusCode::OK, "{}", approved.1);
+                assert_eq!(pending.await.unwrap().0, StatusCode::ACCEPTED);
+            }
+            runtime.close().await;
+            database.remove().await;
+        }
+    }
+}