From d303d0c24e61af566c55a86aae3d3147cc2d362b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 04:32:47 -0600 Subject: [PATCH 1/5] Settle closed-run wait timeout timers without adding history --- .github/workflows/server-parity.yml | 2 + rust/src/runtime/store.rs | 7 +- rust/src/runtime/wait_timeouts.rs | 66 ++++ rust/tests/execution.rs | 3 + rust/tests/execution/closed_wait_timeouts.rs | 300 ++++++++++++++++++ .../closed-wait-timeouts-contract.mjs | 12 + .../closed-wait-timeouts-probe.php | 85 +++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../ServerParity/closed-wait-timeouts.json | 150 +++++++++ ...erParityClosedWaitTimeoutsContractTest.mjs | 46 +++ 11 files changed, 678 insertions(+), 2 deletions(-) create mode 100644 rust/tests/execution/closed_wait_timeouts.rs create mode 100644 scripts/conformance/server-parity/closed-wait-timeouts-contract.mjs create mode 100644 scripts/conformance/server-parity/closed-wait-timeouts-probe.php create mode 100644 tests/Fixtures/ServerParity/closed-wait-timeouts.json create mode 100644 tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index e51f2d4c0..6b61ecd8f 100644 --- a/.github/workflows/server-parity.yml +++ b/.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 \ diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index dafab86b0..9b609e4fb 100644 --- a/rust/src/runtime/store.rs +++ b/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")?, @@ -4495,6 +4497,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?; } diff --git a/rust/src/runtime/wait_timeouts.rs b/rust/src/runtime/wait_timeouts.rs index c6966dc3e..3c08d792e 100644 --- a/rust/src/runtime/wait_timeouts.rs +++ b/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)>> { timeout_for(command, "invalid_signal_timeout") @@ -76,6 +77,71 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { + /// 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, + ) -> 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::>>()?; + // 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(); + 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()); + } + } + 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?; + } + 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?; + Self::query("UPDATE workflow_run_waits SET task_backed=FALSE,task_status='cancelled',updated_at=$1 WHERE workflow_run_id=$2 AND task_id=$3") + .bind(DB::bind_time(at)).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, diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 5fe3195d4..c8638a3cf 100644 --- a/rust/tests/execution.rs +++ b/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; diff --git a/rust/tests/execution/closed_wait_timeouts.rs b/rust/tests/execution/closed_wait_timeouts.rs new file mode 100644 index 000000000..51c0c2abb --- /dev/null +++ b/rust/tests/execution/closed_wait_timeouts.rs @@ -0,0 +1,300 @@ +use super::*; + +fn waits() -> Vec { + vec![ + json!({"type":"open_condition_wait","condition_key":"ready", + "condition_wait_occurrence_id":"closed-wait:0","timeout_seconds":600}), + json!({"type":"open_signal_wait","signal_name":"wake","timeout_seconds":600}), + ] +} + +async fn rows(database: &TestDatabase, table: &str, run: &Value) -> Vec<(String, String)> { + let run = run.as_str().unwrap(); + let query = format!("SELECT id,status FROM {table} WHERE workflow_run_id='{run}' ORDER BY id"); + text_rows(database, &query, &query, &query).await +} + +async fn closed(case: &str) { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "close-worker", json!(["echo"]), json!([])).await; + + // A different live run retains its own pending condition timeout. + let peer = start(&app, "wait-close-peer").await; + let peer_task = poll(&app, "close-worker", "workflow").await; + assert_eq!( + finish_task(&app, &peer_task, json!([waits()[0]])).await.0, + StatusCode::OK + ); + let peer_timers = rows(&database, "workflow_run_timers", &peer["run_id"]).await; + let peer_tasks = database.timer_task_states(&peer["run_id"]).await; + + let run = start(&app, &format!("wait-close-{case}")).await; + let task = poll(&app, "close-worker", "workflow").await; + let mut commands = waits(); + // Ordinary timers retain PHP's existing completion/failure/continue behavior. + commands.push(json!({"type":"start_timer","delay_seconds":600})); + let terminal = match case { + "complete" => Some(json!({"type":"complete_workflow","result":envelope(Payload::Null)})), + "fail" => Some(json!({"type":"fail_workflow","message":"close wait probe"})), + "continue" => Some(json!({"type":"continue_as_new"})), + _ => None, + }; + if let Some(command) = terminal { + commands.push(command); + } + let commands = json!(commands); + let completed = finish_task(&app, &task, commands.clone()).await; + assert_eq!(completed.0, StatusCode::OK, "{}", completed.1); + if matches!(case, "cancel" | "terminate") { + let path = format!( + "/api/workflows/{}/{case}", + run["workflow_id"].as_str().unwrap() + ); + let response = request(&app, "POST", &path, json!({"reason":"close wait probe"})).await; + assert_eq!(response.0, StatusCode::OK, "{}", response.1); + } + + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let mut expected = vec![ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "SignalWaitOpened", + "TimerScheduled", + "TimerScheduled", + ]; + match case { + "complete" => expected.push("WorkflowCompleted"), + "fail" => expected.push("WorkflowFailed"), + "continue" => expected.push("WorkflowContinuedAsNew"), + "cancel" => expected.extend([ + "CancelRequested", + "TimerCancelled", + "TimerCancelled", + "TimerCancelled", + "WorkflowCancelled", + ]), + "terminate" => expected.extend([ + "TerminateRequested", + "TimerCancelled", + "TimerCancelled", + "TimerCancelled", + "WorkflowTerminated", + ]), + _ => unreachable!(), + } + assert_eq!( + history + .as_array() + .unwrap() + .iter() + .map(|event| event["event_type"].as_str().unwrap()) + .collect::>(), + expected + ); + let all_cancelled = matches!(case, "cancel" | "terminate"); + let timers = rows(&database, "workflow_run_timers", &run["run_id"]).await; + assert_eq!(timers.len(), 3); + assert_eq!( + timers + .iter() + .filter(|(_, status)| status == "cancelled") + .count(), + if all_cancelled { 3 } else { 2 } + ); + assert_eq!( + timers + .iter() + .filter(|(_, status)| status == "pending") + .count(), + usize::from(!all_cancelled) + ); + let tasks = database.timer_task_states(&run["run_id"]).await; + assert_eq!(tasks.len(), 3); + assert_eq!( + tasks + .iter() + .filter(|(_, status)| status == "cancelled") + .count(), + if all_cancelled { 3 } else { 2 } + ); + assert_eq!( + tasks.iter().filter(|(_, status)| status == "ready").count(), + usize::from(!all_cancelled) + ); + let lease_query = format!( + "SELECT id,status FROM workflow_tasks WHERE workflow_run_id='{}' AND task_type='timer' AND status='cancelled' AND lease_expires_at IS NOT NULL", + run["run_id"].as_str().unwrap() + ); + assert!( + text_rows(&database, &lease_query, &lease_query, &lease_query) + .await + .is_empty() + ); + assert_eq!( + rows(&database, "workflow_run_timers", &peer["run_id"]).await, + peer_timers + ); + assert_eq!( + database.timer_task_states(&peer["run_id"]).await, + peer_tasks + ); + + // A duplicate completion after a fresh database open cannot recreate or + // re-cancel timers, or add events to either original run. + drop(app); + runtime.close().await; + let cold = database.open().await.unwrap(); + let app = router(cold.clone()); + let duplicate = finish_task(&app, &task, commands).await; + assert_eq!(duplicate.0, StatusCode::OK, "{}", duplicate.1); + assert_eq!(duplicate.1["recorded"], false); + assert_eq!( + run_history(&app, &run["workflow_id"], &run["run_id"]).await, + history + ); + assert_eq!( + rows(&database, "workflow_run_timers", &run["run_id"]).await, + timers + ); + assert_eq!(database.timer_task_states(&run["run_id"]).await, tasks); + assert_eq!( + rows(&database, "workflow_run_timers", &peer["run_id"]).await, + peer_timers + ); + cold.close().await; + database.remove().await; +} + +#[tokio::test] +async fn completion_settles_only_wait_timeouts_and_preserves_php_history() { + closed("complete").await; +} + +#[tokio::test] +async fn failure_settles_only_wait_timeouts_and_preserves_php_history() { + closed("fail").await; +} + +#[tokio::test] +async fn continuation_settles_only_predecessor_wait_timeouts_and_preserves_php_history() { + closed("continue").await; +} + +#[tokio::test] +async fn cancellation_keeps_its_existing_timer_cancellation_history() { + closed("cancel").await; +} + +#[tokio::test] +async fn termination_keeps_its_existing_timer_cancellation_history() { + closed("terminate").await; +} + +async fn later_completion(missing_retained_state: bool) { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register( + &app, + "later-worker", + json!(["echo"]), + json!(["activity-echo"]), + ) + .await; + let run = start(&app, "wait-close-later").await; + let first = poll(&app, "later-worker", "workflow").await; + let mut commands = waits(); + commands.push(json!({"type":"schedule_activity","activity_type":"activity-echo","arguments":envelope(Payload::Array(vec![]))})); + assert_eq!( + finish_task(&app, &first, json!(commands)).await.0, + StatusCode::OK + ); + let activity = poll(&app, "later-worker", "activity").await; + let path = format!( + "/api/worker/activity-tasks/{}/complete", + activity["task_id"].as_str().unwrap() + ); + let done = request( + &app, + "POST", + &path, + json!({"lease_owner":activity["lease_owner"], + "activity_attempt_id":activity["activity_attempt_id"],"result":envelope(Payload::Null)}), + ) + .await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + if missing_retained_state { + // Cleanup uses scheduling history even if a retained projection or + // physical task is unavailable; it must not rebuild deleted rows. + let id = run["run_id"].as_str().unwrap(); + database.execute(sqlx::AssertSqlSafe(format!("DELETE FROM workflow_run_waits WHERE workflow_run_id='{id}' AND kind IN ('condition','signal')"))).await; + database + .execute(sqlx::AssertSqlSafe(format!( + "DELETE FROM workflow_tasks WHERE workflow_run_id='{id}' AND task_type='timer'" + ))) + .await; + } + if !missing_retained_state { + let id = run["run_id"].as_str().unwrap(); + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_tasks SET status='leased',lease_expires_at='2099-01-01 00:00:00' WHERE workflow_run_id='{id}' AND task_type='timer'"))).await; + } + drop(app); + runtime.close().await; + let cold = database.open().await.unwrap(); + let app = router(cold.clone()); + let resumed = poll(&app, "later-worker", "workflow").await; + assert_eq!(resumed["run_id"], run["run_id"]); + let before = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let complete = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]); + let finished = finish_task(&app, &resumed, complete.clone()).await; + assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); + assert_eq!(finished.1["run_status"], "completed"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let events = history.as_array().unwrap(); + assert_eq!( + &events[..events.len() - 1], + before.as_array().unwrap().as_slice() + ); + assert_eq!(events.last().unwrap()["event_type"], "WorkflowCompleted"); + let timers = rows(&database, "workflow_run_timers", &run["run_id"]).await; + assert_eq!(timers.len(), 2); + assert!(timers.iter().all(|(_, state)| state == "cancelled")); + let tasks = database.timer_task_states(&run["run_id"]).await; + assert_eq!(tasks.len(), if missing_retained_state { 0 } else { 2 }); + assert!(tasks.iter().all(|(_, state)| state == "cancelled")); + if missing_retained_state { + let query = format!( + "SELECT id,status FROM workflow_run_waits WHERE workflow_run_id='{}' AND kind IN ('condition','signal')", + run["run_id"].as_str().unwrap() + ); + assert!( + text_rows(&database, &query, &query, &query) + .await + .is_empty() + ); + } + assert_eq!( + finish_task(&app, &resumed, complete).await.1["recorded"], + false + ); + assert_eq!( + run_history(&app, &run["workflow_id"], &run["run_id"]).await, + history + ); + cold.close().await; + database.remove().await; +} + +#[tokio::test] +async fn later_task_closes_original_timeouts_after_database_reopen() { + later_completion(false).await; +} + +#[tokio::test] +async fn retained_projection_or_task_absence_does_not_hide_pending_timeout_rows() { + later_completion(true).await; +} diff --git a/scripts/conformance/server-parity/closed-wait-timeouts-contract.mjs b/scripts/conformance/server-parity/closed-wait-timeouts-contract.mjs new file mode 100644 index 000000000..6c894881a --- /dev/null +++ b/scripts/conformance/server-parity/closed-wait-timeouts-contract.mjs @@ -0,0 +1,12 @@ +import assert from 'node:assert/strict'; + +export function checkClosedWaitTimeouts(fixture, observation) { + const state = observation.closed_wait_timeouts; + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}); + } else { + assert.deepStrictEqual(state, {applicable: true, cases: fixture.closed_wait_timeouts.expected}, + 'terminal waits preserve the published PHP history on every close path'); + } + return {scope: 'closed_wait_timeouts', cases: Object.keys(fixture.closed_wait_timeouts.expected).length}; +} diff --git a/scripts/conformance/server-parity/closed-wait-timeouts-probe.php b/scripts/conformance/server-parity/closed-wait-timeouts-probe.php new file mode 100644 index 000000000..6c698a99e --- /dev/null +++ b/scripts/conformance/server-parity/closed-wait-timeouts-probe.php @@ -0,0 +1,85 @@ +createNamespace($namespace, 'closed wait timeout histories', 30); + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new ClosedWaitHttp(['base_uri' => rtrim($url, '/'), 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $raw = static function (string $method, string $path, ?array $body = null) use ($http, $namespace): array { + $options = ['headers' => array_filter([ + 'Authorization' => getenv('DW_PARITY_TOKEN') ? 'Bearer '.getenv('DW_PARITY_TOKEN') : null, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => '1.20', + ])]; + if ($body !== null) { + $options['json'] = $body === [] ? (object) [] : $body; + } + $response = $http->request($method, $path, $options); + $decoded = json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR); + if ($response->getStatusCode() >= 400) { + throw new RuntimeException('Closed wait timeout '.$method.' '.$path.' HTTP '.$response->getStatusCode() + .': '.($decoded['reason'] ?? $decoded['message'] ?? 'unknown')); + } + + return $decoded; + }; + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $cases = []; + foreach (['condition', 'signal'] as $kind) { + foreach (['complete', 'fail', 'continue', 'cancel', 'terminate'] as $close) { + $id = $workflowId.'-'.$kind.'-'.$close; + $queue = $id.'-queue'; + $worker = $id.'-worker'; + $raw('POST', '/api/worker/register', [ + 'worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'php', + 'supported_workflow_types' => [$definition['workflow_type']], 'supported_activity_types' => [], + 'capabilities' => [], 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18']), + ]); + $run = $raw('POST', '/api/workflows', ['workflow_id' => $id, + 'workflow_type' => $definition['workflow_type'], 'task_queue' => $queue, 'input' => $empty]); + $task = $raw('POST', '/api/worker/workflow-tasks/poll', [ + 'worker_id' => $worker, 'task_queue' => $queue, 'timeout_seconds' => 5, + ])['task'] ?? null; + if (! is_array($task) || ($task['run_id'] ?? null) !== $run['run_id']) { + throw new RuntimeException('Closed wait timeout expected an original run claim.'); + } + $wait = $kind === 'condition' + ? ['type' => 'open_condition_wait', 'condition_key' => 'ready', + 'condition_wait_occurrence_id' => $id.':0', 'timeout_seconds' => $definition['timeout_seconds']] + : ['type' => 'open_signal_wait', 'signal_name' => 'wake', 'timeout_seconds' => $definition['timeout_seconds']]; + $commands = [$wait]; + if ($close === 'complete') { + $commands[] = ['type' => 'complete_workflow', 'result' => $empty]; + } elseif ($close === 'fail') { + $commands[] = ['type' => 'fail_workflow', 'message' => 'close wait probe']; + } elseif ($close === 'continue') { + $commands[] = ['type' => 'continue_as_new']; + } + $raw('POST', '/api/worker/workflow-tasks/'.$task['task_id'].'/complete', [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], 'commands' => $commands, + ]); + if (in_array($close, ['cancel', 'terminate'], true)) { + $raw('POST', '/api/workflows/'.$id.'/'.$close, ['reason' => 'close wait probe']); + } + $path = '/api/workflows/'.$id.'/runs/'.$run['run_id']; + $described = $raw('GET', $path); + $history = $raw('GET', $path.'/history'); + $cases[$kind.'_'.$close] = ['status' => $described['status'], + 'history' => array_column($history['events'] ?? [], 'event_type')]; + } + } + + return ['applicable' => true, 'cases' => $cases]; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 3d924d111..caf1353ee 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -14,6 +14,7 @@ import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkStorageTest} from './storage-test-contract.mjs'; import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; import {checkConcurrentWaits} from './concurrent-waits-contract.mjs'; +import {checkClosedWaitTimeouts} from './closed-wait-timeouts-contract.mjs'; import {checkHistoryExportWaits} from './history-export-waits-contract.mjs'; import {checkContinueTransfers} from './continue-transfers-contract.mjs'; import {checkLargeCommandBatches} from './large-command-batches-contract.mjs'; @@ -194,6 +195,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const storageTest = fixture.storage_test ? checkStorageTest(fixture, observation, workflowId) : null; const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null; const concurrentWaits = fixture.concurrent_waits ? checkConcurrentWaits(fixture, observation) : null; + const closedWaitTimeouts = fixture.closed_wait_timeouts ? checkClosedWaitTimeouts(fixture, observation) : null; const historyExportWaits = fixture.history_export_waits ? checkHistoryExportWaits(fixture, observation) : null; const continueTransfers = fixture.continue_transfers ? checkContinueTransfers(fixture, observation) : null; const largeCommandBatches = fixture.large_command_batches ? checkLargeCommandBatches(fixture, observation) : null; @@ -792,6 +794,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(storageTest ? {storage_test: storageTest} : {}), ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}), ...(concurrentWaits ? {concurrent_waits: concurrentWaits} : {}), + ...(closedWaitTimeouts ? {closed_wait_timeouts: closedWaitTimeouts} : {}), ...(historyExportWaits ? {history_export_waits: historyExportWaits} : {}), ...(continueTransfers ? {continue_transfers: continueTransfers} : {}), ...(largeCommandBatches ? {large_command_batches: largeCommandBatches} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index efcc62855..3fd41d846 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -432,6 +432,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $storageTest = isset($fixture['storage_test']) ? finishHttpStorageTest($client, $fixture, $workflowId, $url) : null; $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null; $concurrentWaits = isset($fixture['concurrent_waits']) ? finishHttpConcurrentWaits($client, $fixture, $workflowId, $url) : null; + $closedWaitTimeouts = isset($fixture['closed_wait_timeouts']) ? finishHttpClosedWaitTimeouts($client, $fixture, $workflowId, $url) : null; $historyExportWaits = isset($fixture['history_export_waits']) ? finishHttpHistoryExportWaits($client, $fixture, $workflowId, $url) : null; $continueTransfers = isset($fixture['continue_transfers']) ? finishHttpContinueTransfers($client, $fixture, $workflowId, $url) : null; $largeCommandBatches = isset($fixture['large_command_batches']) ? finishHttpLargeCommandBatches($client, $fixture, $workflowId, $url) : null; @@ -583,6 +584,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'storage_test' => $storageTest, 'activity_lease_renewal' => $activityLeaseRenewal, 'concurrent_waits' => $concurrentWaits, + 'closed_wait_timeouts' => $closedWaitTimeouts, 'history_export_waits' => $historyExportWaits, 'continue_transfers' => $continueTransfers, 'large_command_batches' => $largeCommandBatches, @@ -940,6 +942,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/storage-test-probe.php'; require __DIR__.'/activity-lease-renewal-probe.php'; require __DIR__.'/concurrent-waits-probe.php'; + require __DIR__.'/closed-wait-timeouts-probe.php'; require __DIR__.'/history-export-waits-probe.php'; require __DIR__.'/continue-transfers-probe.php'; require __DIR__.'/large-command-batches-probe.php'; @@ -1150,6 +1153,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['concurrent_waits'])) { $observation['concurrent_waits'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } + if ($mode === 'embedded' && isset($fixture['closed_wait_timeouts'])) { + $observation['closed_wait_timeouts'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; + } if ($mode === 'embedded' && isset($fixture['history_export_waits'])) { $observation['history_export_waits'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } diff --git a/tests/Fixtures/ServerParity/closed-wait-timeouts.json b/tests/Fixtures/ServerParity/closed-wait-timeouts.json new file mode 100644 index 000000000..fdeb1bb7a --- /dev/null +++ b/tests/Fixtures/ServerParity/closed-wait-timeouts.json @@ -0,0 +1,150 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": { + "message": "concurrent waits λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "concurrent waits λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "id": "closed-wait-timeouts", + "contract": "Published PHP Server 2.11.8 history for a timed condition or signal wait followed by complete, fail, continue-as-new, cancel or terminate. Ordinary complete/fail/continue add no TimerCancelled or wait-resolution event; cancel/terminate cancel the wait timeout before their terminal event. Native additionally settles its physical pending timeout timer and ready/leased timer task on every close, without changing those histories. Raw worker HTTP uses a fresh namespace; embedded applications have no worker protocol. Physical cleanup is asserted separately by cross-backend native execution tests.", + "closed_wait_timeouts": { + "workflow_type": "parity.v1.close_wait", + "empty_blob": "wwHioz3/VYAiNwwA", + "timeout_seconds": 600, + "expected": { + "condition_complete": { + "status": "completed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "WorkflowCompleted" + ] + }, + "condition_fail": { + "status": "failed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "WorkflowFailed" + ] + }, + "condition_continue": { + "status": "completed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "WorkflowContinuedAsNew" + ] + }, + "condition_cancel": { + "status": "cancelled", + "history": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "CancelRequested", + "TimerCancelled", + "WorkflowCancelled" + ] + }, + "condition_terminate": { + "status": "terminated", + "history": [ + "StartAccepted", + "WorkflowStarted", + "ConditionWaitOpened", + "TimerScheduled", + "TerminateRequested", + "TimerCancelled", + "WorkflowTerminated" + ] + }, + "signal_complete": { + "status": "completed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "TimerScheduled", + "WorkflowCompleted" + ] + }, + "signal_fail": { + "status": "failed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "TimerScheduled", + "WorkflowFailed" + ] + }, + "signal_continue": { + "status": "completed", + "history": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "TimerScheduled", + "WorkflowContinuedAsNew" + ] + }, + "signal_cancel": { + "status": "cancelled", + "history": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "TimerScheduled", + "CancelRequested", + "TimerCancelled", + "WorkflowCancelled" + ] + }, + "signal_terminate": { + "status": "terminated", + "history": [ + "StartAccepted", + "WorkflowStarted", + "SignalWaitOpened", + "TimerScheduled", + "TerminateRequested", + "TimerCancelled", + "WorkflowTerminated" + ] + } + } + } +} diff --git a/tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs b/tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs new file mode 100644 index 000000000..98d6587eb --- /dev/null +++ b/tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs @@ -0,0 +1,46 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkClosedWaitTimeouts} from '../../scripts/conformance/server-parity/closed-wait-timeouts-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/closed-wait-timeouts.json', import.meta.url), 'utf8')); +const observed = () => ({mode: 'http', closed_wait_timeouts: {applicable: true, + cases: structuredClone(fixture.closed_wait_timeouts.expected)}}); + +test('all ten published PHP close histories pass', () => { + assert.deepStrictEqual(checkClosedWaitTimeouts(fixture, observed()), {scope: 'closed_wait_timeouts', cases: 10}); +}); + +test('embedded explicitly excludes the HTTP worker contract', () => { + assert.deepStrictEqual(checkClosedWaitTimeouts(fixture, + {mode: 'embedded', closed_wait_timeouts: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'closed_wait_timeouts', cases: 10}); + assert.throws(() => checkClosedWaitTimeouts(fixture, {mode: 'embedded', closed_wait_timeouts: {applicable: true}})); +}); + +for (const kind of ['condition', 'signal']) { + for (const close of ['complete', 'fail', 'continue', 'cancel', 'terminate']) { + test(`rejects an altered ${kind} ${close} history`, () => { + const observation = observed(); + const events = observation.closed_wait_timeouts.cases[`${kind}_${close}`].history; + if (['cancel', 'terminate'].includes(close)) { + events.splice(events.indexOf('TimerCancelled'), 1); + } else { + events.splice(-1, 0, 'TimerCancelled'); + } + assert.throws(() => checkClosedWaitTimeouts(fixture, observation)); + }); + } +} + +test('a missing close path cannot masquerade as a complete comparison', () => { + const observation = observed(); + delete observation.closed_wait_timeouts.cases.signal_continue; + assert.throws(() => checkClosedWaitTimeouts(fixture, observation)); +}); + +test('a workflow result with the wrong terminal state fails', () => { + const observation = observed(); + observation.closed_wait_timeouts.cases.condition_fail.status = 'completed'; + assert.throws(() => checkClosedWaitTimeouts(fixture, observation)); +}); From 84f9faa9c71c2081a5153357bc79b7551a048679 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 04:48:56 -0600 Subject: [PATCH 2/5] Keep the leased-timeout test inside the database timestamp range --- rust/tests/execution/closed_wait_timeouts.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/tests/execution/closed_wait_timeouts.rs b/rust/tests/execution/closed_wait_timeouts.rs index 51c0c2abb..60fe1cb8a 100644 --- a/rust/tests/execution/closed_wait_timeouts.rs +++ b/rust/tests/execution/closed_wait_timeouts.rs @@ -240,7 +240,10 @@ async fn later_completion(missing_retained_state: bool) { } if !missing_retained_state { let id = run["run_id"].as_str().unwrap(); - database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_tasks SET status='leased',lease_expires_at='2099-01-01 00:00:00' WHERE workflow_run_id='{id}' AND task_type='timer'"))).await; + let lease_until = (chrono::Utc::now() + chrono::Duration::hours(1)) + .format("%Y-%m-%d %H:%M:%S%.6f") + .to_string(); + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_tasks SET status='leased',lease_expires_at='{lease_until}' WHERE workflow_run_id='{id}' AND task_type='timer'"))).await; } drop(app); runtime.close().await; From d1de3def6c2de3f7aaa351b2c0c44fe2b62ada73 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:23:55 -0600 Subject: [PATCH 3/5] test: cover surviving waits without physical timeout tasks --- rust/tests/execution/closed_wait_timeouts.rs | 52 +++++++++++++++++--- 1 file changed, 45 insertions(+), 7 deletions(-) diff --git a/rust/tests/execution/closed_wait_timeouts.rs b/rust/tests/execution/closed_wait_timeouts.rs index 60fe1cb8a..d669e4b0d 100644 --- a/rust/tests/execution/closed_wait_timeouts.rs +++ b/rust/tests/execution/closed_wait_timeouts.rs @@ -194,7 +194,7 @@ async fn termination_keeps_its_existing_timer_cancellation_history() { closed("terminate").await; } -async fn later_completion(missing_retained_state: bool) { +async fn later_completion(missing_projection: bool, missing_task: bool) { let database = TestDatabase::new().await; let runtime = database.open().await.unwrap(); let app = router(runtime.clone()); @@ -227,18 +227,21 @@ async fn later_completion(missing_retained_state: bool) { ) .await; assert_eq!(done.0, StatusCode::OK, "{}", done.1); - if missing_retained_state { + if missing_projection { // Cleanup uses scheduling history even if a retained projection or // physical task is unavailable; it must not rebuild deleted rows. let id = run["run_id"].as_str().unwrap(); database.execute(sqlx::AssertSqlSafe(format!("DELETE FROM workflow_run_waits WHERE workflow_run_id='{id}' AND kind IN ('condition','signal')"))).await; + } + if missing_task { + let id = run["run_id"].as_str().unwrap(); database .execute(sqlx::AssertSqlSafe(format!( "DELETE FROM workflow_tasks WHERE workflow_run_id='{id}' AND task_type='timer'" ))) .await; } - if !missing_retained_state { + if !missing_task { let id = run["run_id"].as_str().unwrap(); let lease_until = (chrono::Utc::now() + chrono::Duration::hours(1)) .format("%Y-%m-%d %H:%M:%S%.6f") @@ -252,6 +255,19 @@ async fn later_completion(missing_retained_state: bool) { let resumed = poll(&app, "later-worker", "workflow").await; assert_eq!(resumed["run_id"], run["run_id"]); let before = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + if missing_task && !missing_projection { + let id = run["run_id"].as_str().unwrap(); + let surviving = format!( + "SELECT id,task_status FROM workflow_run_waits WHERE workflow_run_id='{id}' AND kind IN ('condition','signal') AND task_backed=TRUE AND task_status='ready'" + ); + assert_eq!( + text_rows(&database, &surviving, &surviving, &surviving) + .await + .len(), + 2, + "both task-backed projections survive without physical timer tasks before close" + ); + } let complete = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]); let finished = finish_task(&app, &resumed, complete.clone()).await; assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); @@ -267,13 +283,30 @@ async fn later_completion(missing_retained_state: bool) { assert_eq!(timers.len(), 2); assert!(timers.iter().all(|(_, state)| state == "cancelled")); let tasks = database.timer_task_states(&run["run_id"]).await; - assert_eq!(tasks.len(), if missing_retained_state { 0 } else { 2 }); + assert_eq!(tasks.len(), if missing_task { 0 } else { 2 }); assert!(tasks.iter().all(|(_, state)| state == "cancelled")); - if missing_retained_state { + if missing_projection { let query = format!( "SELECT id,status FROM workflow_run_waits WHERE workflow_run_id='{}' AND kind IN ('condition','signal')", run["run_id"].as_str().unwrap() ); + } else { + let id = run["run_id"].as_str().unwrap(); + let waits = format!( + "SELECT id,status FROM workflow_run_waits WHERE workflow_run_id='{id}' AND kind IN ('condition','signal')" + ); + let states = text_rows(&database, &waits, &waits, &waits).await; + assert_eq!(states.len(), 2); + assert!(states.iter().all(|(_, status)| status == "cancelled")); + let backed = format!( + "SELECT id,task_status FROM workflow_run_waits WHERE workflow_run_id='{id}' AND kind IN ('condition','signal') AND (task_backed=TRUE OR task_status<>'cancelled' OR task_status IS NULL)" + ); + assert!( + text_rows(&database, &backed, &backed, &backed) + .await + .is_empty(), + "terminal waits retain no ready/leased task association" + ); assert!( text_rows(&database, &query, &query, &query) .await @@ -294,10 +327,15 @@ async fn later_completion(missing_retained_state: bool) { #[tokio::test] async fn later_task_closes_original_timeouts_after_database_reopen() { - later_completion(false).await; + later_completion(false, false).await; } #[tokio::test] async fn retained_projection_or_task_absence_does_not_hide_pending_timeout_rows() { - later_completion(true).await; + later_completion(true, true).await; +} + +#[tokio::test] +async fn absent_timer_tasks_still_clear_surviving_wait_projection_associations() { + later_completion(false, true).await; } From c2f374bde09a14b4ee4009b786ee13263c78735c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:24:47 -0600 Subject: [PATCH 4/5] test: scope the absent-projection assertion correctly --- rust/tests/execution/closed_wait_timeouts.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/rust/tests/execution/closed_wait_timeouts.rs b/rust/tests/execution/closed_wait_timeouts.rs index d669e4b0d..664e9134a 100644 --- a/rust/tests/execution/closed_wait_timeouts.rs +++ b/rust/tests/execution/closed_wait_timeouts.rs @@ -290,6 +290,11 @@ async fn later_completion(missing_projection: bool, missing_task: bool) { "SELECT id,status FROM workflow_run_waits WHERE workflow_run_id='{}' AND kind IN ('condition','signal')", run["run_id"].as_str().unwrap() ); + assert!( + text_rows(&database, &query, &query, &query) + .await + .is_empty() + ); } else { let id = run["run_id"].as_str().unwrap(); let waits = format!( @@ -307,11 +312,6 @@ async fn later_completion(missing_projection: bool, missing_task: bool) { .is_empty(), "terminal waits retain no ready/leased task association" ); - assert!( - text_rows(&database, &query, &query, &query) - .await - .is_empty() - ); } assert_eq!( finish_task(&app, &resumed, complete).await.1["recorded"], From 7b1818d6b4aac48307d3c19d26c581065cfc797b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:25:23 -0600 Subject: [PATCH 5/5] fix: clear closed wait projections without timeout tasks --- rust/src/runtime/wait_timeouts.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/rust/src/runtime/wait_timeouts.rs b/rust/src/runtime/wait_timeouts.rs index 3c08d792e..fa2cbb297 100644 --- a/rust/src/runtime/wait_timeouts.rs +++ b/rust/src/runtime/wait_timeouts.rs @@ -105,6 +105,7 @@ where 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!( @@ -115,6 +116,14 @@ where .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() { @@ -124,6 +133,13 @@ where 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 { @@ -135,8 +151,6 @@ where 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?; - Self::query("UPDATE workflow_run_waits SET task_backed=FALSE,task_status='cancelled',updated_at=$1 WHERE workflow_run_id=$2 AND task_id=$3") - .bind(DB::bind_time(at)).bind(run_id).bind(&task_id).execute(&mut **tx).await?; } } Ok(())