Make unresolved condition-wait opens idempotent by occurrence ID

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).

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

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));

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()

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<Value> {
+    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;

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