diff --git a/docs/contracts/condition-wait-occurrences.md b/docs/contracts/condition-wait-occurrences.md new file mode 100644 index 000000000..7432caa7d --- /dev/null +++ b/docs/contracts/condition-wait-occurrences.md @@ -0,0 +1,31 @@ +# Idempotent condition-wait opens + +Native Server advertises `condition_wait_idempotent_reopen: true` in cluster +capabilities and worker `server_capabilities` when it recognizes repeated +`open_condition_wait` commands by occurrence ID. + +If a command carries a nonempty `condition_wait_occurrence_id` matching an +unresolved condition wait in the same run, Server keeps the existing wait. It +records no second `ConditionWaitOpened`, consumes no new authored sequence and +schedules no second timeout. The original wait identity, key, fingerprint and +deadline stay intact. Other commands in the completion still apply normally. + +An occurrence is unresolved while its wait is open and its timeout has not +fired. After satisfaction, timeout or cancellation, another open with the same +occurrence ID creates a new physical wait. A false-predicate wake therefore +keeps the existing reopen behavior. SDKs retain the logical occurrence ID and +send the remaining timeout when they reopen after such a wake. + +Occurrence IDs belong to one logical wait call. Use a deterministic per-run +ordinal in workflow-code order, reuse it on replay and false-predicate wakes, +and allocate a different ID for a separate wait call. The ID is independent of +the condition key, predicate fingerprint, worker process and task attempt. + +Absent, null and empty occurrence IDs preserve the existing per-open behavior. +Distinct occurrence IDs can share a key or fingerprint without being combined. +Signal waits are unaffected. Published PHP Server does not advertise this +capability and keeps its per-open behavior. + +This capability does not settle unresolved waits' timers when a run closes; +that is the second change tracked in +[Server #1015](https://github.com/durable-workflow/server/issues/1015). diff --git a/docs/server-reference.md b/docs/server-reference.md index bf7a2912d..63696884b 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -937,6 +937,9 @@ Memo upserts and explicit typed search-attribute updates additionally require the matching registered worker capability. Durable condition-wait occurrence identity requires protocol `1.17`; older workers remain eligible for workflows whose replay history does not contain that identity. +Native additionally advertises `condition_wait_idempotent_reopen` when opening +an already-unresolved occurrence preserves its existing wait and deadline. +See [idempotent condition-wait opens](contracts/condition-wait-occurrences.md). Long-poll wake-ups use short-lived cache-backed signal keys plus periodic reprobes. Multi-node deployments therefore need a shared cache backend for diff --git a/rust/src/runtime/cluster_info.rs b/rust/src/runtime/cluster_info.rs index fa7f787ba..f60e9a13a 100644 --- a/rust/src/runtime/cluster_info.rs +++ b/rust/src/runtime/cluster_info.rs @@ -62,6 +62,7 @@ pub(super) fn capabilities() -> Value { "child_workflow_retry_policy", "child_workflow_timeouts", "child_workflow_start_turn", + "condition_wait_idempotent_reopen", "parent_close_policy", "non_retryable_failures", "worker_sessions", @@ -209,9 +210,10 @@ mod tests { "workflow_update_runtime_contract", "waterline_v1_projection", "embedded_v2_import"] { assert!(caps.get(name).is_some(), "{name}"); } - // PHP's names plus native's child start turn (Server #966). - assert_eq!(caps.as_object().unwrap().len(), 57); + // PHP's names plus native's child turn and occurrence-id reopen. + assert_eq!(caps.as_object().unwrap().len(), 58); assert_eq!(caps["child_workflow_start_turn"], json!(true)); + assert_eq!(caps["condition_wait_idempotent_reopen"], json!(true)); assert_eq!(caps["response_compression"], json!([])); assert_eq!(caps["timer_runtime_contract"], json!(false)); assert_eq!(caps["principal_attribution_contract"], json!(true)); diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index dafab86b0..46288e270 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3553,10 +3553,23 @@ where } else if command["type"] == "open_condition_wait" || command["type"] == "open_signal_wait" { - // A run may hold any number of open waits - // (`applyOpenSignalWait`, `applyOpenConditionWait`). - sequence += 1; let direct = command["type"] == "open_signal_wait"; + // A queued task can replay an occurrence another task already + // opened. Keep that unresolved wait and its original deadline. + // Keys and fingerprints can name distinct concurrent waits. + if !direct + && let Some(occurrence) = command["condition_wait_occurrence_id"] + .as_str() + .filter(|value| !value.is_empty()) + && Self::waits(tx, &run_id).await?.conditions.iter().any(|wait| { + wait.advanceable() && wait.occurrence() == Some(occurrence) + }) + { + return Ok(()); + } + // Legacy opens and resolved occurrences keep PHP's per-open + // behavior, including several waits with the same key. + sequence += 1; let mut payload = command.clone(); payload.as_object_mut().unwrap().remove("type"); // Match PHP's optional occurrence payload: retain authored @@ -5212,7 +5225,7 @@ pub(crate) fn capabilities() -> Value { "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(), "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true, "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true, - "child_workflow_start_turn": true}); + "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true}); value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest(); value["supported_workflow_task_commands"] .as_array_mut() diff --git a/rust/tests/execution/condition_occurrences.rs b/rust/tests/execution/condition_occurrences.rs index 172cc3d8c..aa8b341b4 100644 --- a/rust/tests/execution/condition_occurrences.rs +++ b/rust/tests/execution/condition_occurrences.rs @@ -21,6 +21,173 @@ async fn register_signal_worker(app: &Router, worker: &str) { assert_eq!(response.0, StatusCode::CREATED, "{}", response.1); } +fn occurrence_events(history: &Value, kind: &str) -> Vec { + history.as_array().unwrap().iter() + .filter(|event| event["event_type"] == kind) + .map(|event| event["payload"].clone()).collect() +} + +async fn register_repeat_worker(app: &Router, worker: &str) { + let (status, body) = request(app, "POST", "/api/worker/register", json!({ + "worker_id":worker,"task_queue":"test","runtime":"rust", + "supported_workflow_types":["echo"],"supported_activity_types":["activity-echo"], + "workflow_command_contracts":{"echo":{"signals":["payload"], + "signal_contracts":[{"name":"payload","parameters":[]}]}} + })).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + assert_eq!(body["server_capabilities"]["condition_wait_idempotent_reopen"], true); +} + +#[tokio::test] +async fn another_task_reopening_an_unresolved_occurrence_keeps_its_wait_and_deadline() { + let database = TestDatabase::new().await; + let first = database.open().await.unwrap(); + let app = router(first.clone()); + register_repeat_worker(&app, "occurrence-opener").await; + let (status, info) = request(&app, "GET", "/api/cluster/info", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{info}"); + assert_eq!(info["capabilities"]["condition_wait_idempotent_reopen"], true); + let run = start(&app, "duplicate-open-run").await; + let original = poll(&app, "occurrence-opener", "workflow").await; + let mut open = condition(json!("rust:condition-wait:0")); + open["timeout_seconds"] = json!(600); + let (status, body) = finish_task(&app, &original, json!([ + open, + {"type":"schedule_activity","activity_type":"activity-echo", + "arguments":envelope(Payload::Array(vec![]))} + ])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let prefix = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let wait = occurrence_events(&prefix, "ConditionWaitOpened"); + let timer = occurrence_events(&prefix, "TimerScheduled"); + assert_eq!((wait.len(), timer.len()), (1, 1)); + + // Activity completion creates another workflow task without satisfying + // the condition. A fresh process must recognize its durable occurrence. + let activity = poll(&app, "occurrence-opener", "activity").await; + let (status, body) = request(&app, "POST", &format!("/api/worker/activity-tasks/{}/complete", + activity["task_id"].as_str().unwrap()), json!({ + "lease_owner":activity["lease_owner"],"activity_attempt_id":activity["activity_attempt_id"], + "result":envelope(Payload::Long(7)) + })).await; + assert_eq!(status, StatusCode::OK, "{body}"); + drop(app); + first.close().await; + let cold = database.open().await.unwrap(); + let app = router(cold.clone()); + register_repeat_worker(&app, "occurrence-replayer").await; + let next = poll(&app, "occurrence-replayer", "workflow").await; + assert_ne!(next["task_id"], original["task_id"]); + let before = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + assert!(occurrence_events(&before, "ConditionWaitSatisfied").is_empty()); + let mut duplicate = condition(json!("rust:condition-wait:0")); + duplicate["timeout_seconds"] = json!(1); + duplicate["condition_key"] = json!("changed-key"); + duplicate["condition_definition_fingerprint"] = json!("changed-fingerprint"); + let (status, body) = finish_task(&app, &next, json!([ + duplicate, + {"type":"record_version_marker","change_id":"after-repeat","version":1, + "min_supported":1,"max_supported":1} + ])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["created_task_ids"], json!([])); + let after = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + assert_eq!(occurrence_events(&after, "ConditionWaitOpened"), wait); + assert_eq!(occurrence_events(&after, "TimerScheduled"), timer); + assert_eq!(after.as_array().unwrap().len(), before.as_array().unwrap().len() + 1); + assert_eq!(occurrence_events(&after, "VersionMarkerRecorded")[0]["sequence"], 3); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_run_timers WHERE status='pending'").await, 1); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_tasks WHERE task_type='timer' AND status='ready'").await, 1); + + let (status, body) = request(&app, "POST", "/api/workflows/duplicate-open-run/signal/payload", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let resumed = poll(&app, "occurrence-replayer", "workflow").await; + let (status, body) = finish_task(&app, &resumed, json!([ + {"type":"complete_workflow","result":envelope(Payload::Long(7))} + ])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["run_status"], "completed"); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_run_timers WHERE status='pending'").await, 0); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_tasks WHERE task_type='timer' AND status='ready'").await, 0); + cold.close().await; + database.remove().await; +} + +#[tokio::test] +async fn an_occurrence_is_idempotent_within_one_batch_without_combining_other_occurrences() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "batch-occurrences", json!(["echo"]), json!([])).await; + let run = start(&app, "batch-occurrences").await; + let task = poll(&app, "batch-occurrences", "workflow").await; + let first = condition(json!("one")); + let (status, body) = finish_task(&app, &task, json!([ + first.clone(), first, condition(json!("two")) + ])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let waits = occurrence_events(&history, "ConditionWaitOpened"); + assert_eq!(waits.len(), 2); + assert_eq!((waits[0]["sequence"].clone(), waits[1]["sequence"].clone()), (json!(1), json!(2))); + assert_eq!((waits[0]["condition_wait_occurrence_id"].clone(), waits[1]["condition_wait_occurrence_id"].clone()), + (json!("one"), json!("two"))); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn repeated_legacy_opens_without_an_occurrence_keep_distinct_waits_and_timers() { + for occurrence in [None, Some(Value::Null), Some(json!(""))] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "legacy-opens", json!(["echo"]), json!([])).await; + let run = start(&app, "legacy-opens").await; + let task = poll(&app, "legacy-opens", "workflow").await; + let mut command = condition(occurrence.clone().unwrap_or(Value::Null)); + if occurrence.is_none() { + command.as_object_mut().unwrap().remove("condition_wait_occurrence_id"); + } + command["timeout_seconds"] = json!(600); + let (status, body) = finish_task(&app, &task, json!([command.clone(), command])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + assert_eq!(occurrence_events(&history, "ConditionWaitOpened").len(), 2); + assert_eq!(occurrence_events(&history, "TimerScheduled").len(), 2); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_run_timers WHERE status='pending'").await, 2); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn an_occurrence_whose_timeout_fired_can_open_a_new_physical_wait() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "expired-occurrence", json!(["echo"]), json!([])).await; + let run = start(&app, "expired-occurrence").await; + let task = poll(&app, "expired-occurrence", "workflow").await; + let mut command = condition(json!("same")); + command["timeout_seconds"] = json!(0); + let (status, body) = finish_task(&app, &task, json!([command.clone()])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let next = poll(&app, "expired-occurrence", "workflow").await; + assert!(!next.is_null()); + command["timeout_seconds"] = json!(600); + let (status, body) = finish_task(&app, &next, json!([command])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let waits = occurrence_events(&history, "ConditionWaitOpened"); + assert_eq!(waits.len(), 2); + assert_ne!(waits[0]["condition_wait_id"], waits[1]["condition_wait_id"]); + assert_eq!(occurrence_events(&history, "TimerFired").len(), 1); + assert_eq!(task_retry::rows(&database, "SELECT COUNT(*) FROM workflow_run_timers WHERE status='pending'").await, 1); + runtime.close().await; + database.remove().await; +} + #[tokio::test] async fn authored_occurrences_survive_restart_and_completion_fences() { let database = TestDatabase::new().await; diff --git a/tests/Fixtures/ServerParity/cluster-capabilities.json b/tests/Fixtures/ServerParity/cluster-capabilities.json index 9b63ef0ef..002021131 100644 --- a/tests/Fixtures/ServerParity/cluster-capabilities.json +++ b/tests/Fixtures/ServerParity/cluster-capabilities.json @@ -368,6 +368,10 @@ } }, "native_additions": { + "capabilities.condition_wait_idempotent_reopen": { + "value": true, + "reason": "Native preserves an unresolved condition wait when a later open repeats its nonempty condition_wait_occurrence_id, keeping the original sequence and deadline (Server #1015). Published PHP retains per-open behavior." + }, "capabilities.child_workflow_start_turn": { "value": true, "reason": "Native serves start_child_workflow await: \"started\", a parent turn when the child start commits (Server #966); published PHP 2.x has no such mode."