Settle wait timeout timers when their run closes

.github/workflows/server-parity.yml

@@ -61,6 +61,7 @@ on:
       - tests/Unit/ServerParityOperatorMetricsContractTest.mjs
       - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs
       - tests/Unit/ServerParityConcurrentWaitsContractTest.mjs
+      - tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs
       - tests/Unit/ServerParityHistoryExportWaitsContractTest.mjs
       - tests/Unit/ServerParityContinueTransfersContractTest.mjs
       - tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs
@@ -2267,6 +2268,7 @@ jobs:
               tests/Unit/ServerParityOperatorMetricsContractTest.mjs \
               tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs \
               tests/Unit/ServerParityConcurrentWaitsContractTest.mjs \
+              tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs \
               tests/Unit/ServerParityHistoryExportWaitsContractTest.mjs \
               tests/Unit/ServerParityContinueTransfersContractTest.mjs \
               tests/Unit/ServerParityChildDescriptionsContractTest.mjs \

rust/src/runtime/store.rs

@@ -3243,8 +3243,10 @@ where
                     Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1").bind(&run_id).execute(&mut *tx).await?;
                 } else {
                     // Published PHP completes a run whatever is still open. Its
-                    // activities, timers and waits stay as they are, and its
-                    // open children follow their parent-close policies.
+                    // ordinary activities and timers stay as they are, and
+                    // open children follow their parent-close policies. The
+                    // terminal event settles physical wait timeout rows while
+                    // retaining PHP's history.
                     let output = match external.get(&index) {
                         Some(prepared) => prepared.stored(),
                         None => envelope(command, "result")?,
@@ -4508,6 +4510,7 @@ where
         Self::project_child_wait_event(tx, run_id, event_type, &payload, recorded_at).await?;
         Self::project_update_wait_event(tx, run_id, event_type, &payload, recorded_at).await?;
         if let Some(reason) = closed_reason(event_type) {
+            Self::settle_closed_wait_timeouts(tx, run_id, recorded_at).await?;
             let task = payload.get("task").cloned();
             Box::pin(Self::reject_received_signals(tx, run_id, reason, task_id, task)).await?;
         }

rust/src/runtime/wait_timeouts.rs

@@ -10,6 +10,7 @@ use axum::http::StatusCode;
 use chrono::{DateTime, Utc};
 use serde_json::{Value, json};
 use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
+use std::collections::BTreeSet;
 
 pub(super) fn timeout(command: &Value) -> Result<Option<(i64, DateTime<Utc>)>> {
     timeout_for(command, "invalid_signal_timeout")
@@ -76,6 +77,85 @@ where
     for<'r> i64: Decode<'r, DB>,
     usize: ColumnIndex<DB::Row>,
 {
+    /// A terminal run cannot fire its remaining wait timeouts. Settle their
+    /// physical rows without adding the cancellation events that published PHP
+    /// omits for ordinary completion, failure and continuation. Authored timers
+    /// retain their existing close behavior.
+    pub(super) async fn settle_closed_wait_timeouts(
+        tx: &mut Transaction<'_, DB>,
+        run_id: &str,
+        at: DateTime<Utc>,
+    ) -> Result<()> {
+        let pending = Self::query(
+            "SELECT id FROM workflow_run_timers WHERE workflow_run_id=$1 AND status='pending'",
+        )
+        .bind(run_id)
+        .fetch_all(&mut **tx)
+        .await?;
+        if pending.is_empty() {
+            return Ok(());
+        }
+        let pending = pending
+            .iter()
+            .map(|row| DB::string(row, "id"))
+            .collect::<Result<BTreeSet<_>>>()?;
+        // Use canonical scheduling history, including when a retained wait
+        // projection or its physical timer task is absent. Read it once for the
+        // run rather than once for each pending timer.
+        let scheduled = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='TimerScheduled'")
+            .bind(run_id).fetch_all(&mut **tx).await?;
+        let mut cancelled = BTreeSet::new();
+        let mut closed_waits = BTreeSet::new();
+        for event in scheduled {
+            let payload = DB::document_row(&event, "payload")?;
+            if matches!(
+                payload["timer_kind"].as_str(),
+                Some("condition_timeout" | "signal_timeout")
+            ) && let Some(timer_id) = payload["timer_id"]
+                .as_str()
+                .filter(|id| pending.contains(*id))
+            {
+                cancelled.insert(timer_id.to_owned());
+                let (kind, field) = if payload["timer_kind"] == "condition_timeout" {
+                    ("condition", "condition_wait_id")
+                } else {
+                    ("signal", "signal_wait_id")
+                };
+                if let Some(wait_id) = payload[field].as_str().filter(|id| !id.is_empty()) {
+                    closed_waits.insert((kind, wait_id.to_owned(), timer_id.to_owned()));
+                }
+            }
+        }
+        if cancelled.is_empty() {
+            return Ok(());
+        }
+        for timer_id in &cancelled {
+            Self::query("UPDATE workflow_run_timers SET status='cancelled' WHERE workflow_run_id=$1 AND id=$2 AND status='pending'")
+                .bind(run_id).bind(timer_id).execute(&mut **tx).await?;
+        }
+        // A retained wait may outlive its physical timer task. Match the
+        // canonical wait and timer identities directly; do not recreate a
+        // missing projection or depend on a task inventory row to clear it.
+        for (kind, wait_id, timer_id) in closed_waits {
+            Self::query("UPDATE workflow_run_waits SET task_backed=FALSE,task_status='cancelled',updated_at=$1 WHERE workflow_run_id=$2 AND kind=$3 AND wait_id=$4 AND resume_source_kind='timer' AND resume_source_id=$5")
+                .bind(DB::bind_time(at)).bind(run_id).bind(kind).bind(wait_id).bind(timer_id).execute(&mut **tx).await?;
+        }
+        let tasks = Self::query("SELECT id,payload FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='timer' AND status IN ('ready','leased')")
+            .bind(run_id).fetch_all(&mut **tx).await?;
+        for task in tasks {
+            let payload = DB::document_row(&task, "payload")?;
+            if payload["timer_id"]
+                .as_str()
+                .is_some_and(|id| cancelled.contains(id))
+            {
+                let task_id = DB::string(&task, "id")?;
+                Self::query("UPDATE workflow_tasks SET status='cancelled',lease_expires_at=NULL WHERE workflow_run_id=$1 AND id=$2")
+                    .bind(run_id).bind(&task_id).execute(&mut **tx).await?;
+            }
+        }
+        Ok(())
+    }
+
     pub(super) async fn schedule_wait_timeout(
         tx: &mut Transaction<'_, DB>,
         run: &DB::Row,

rust/tests/execution.rs

@@ -285,6 +285,9 @@ mod timed_signal_waits;
 #[path = "execution/timed_condition_waits.rs"]
 mod timed_condition_waits;
 
+#[path = "execution/closed_wait_timeouts.rs"]
+mod closed_wait_timeouts;
+
 #[path = "execution/parallel_all.rs"]
 mod parallel_all;