Replay condition waits the Server reopened while still open

CHANGELOG.md

@@ -2,6 +2,20 @@
 
 ## Unreleased
 
+- Replay condition-wait histories in which the Server recorded a second open
+  of a wait that was still open (#191, #192). A wait is now linked to the
+  same logical wait by its state when the next open was recorded: if it was
+  still open, the next open is a repeated open. So an orphaned duplicate whose
+  timer fires later no longer fails replay with
+  `condition_wait_reopened_after_timeout` (the sdk-php#283 shape). On PHP
+  Server 2.x, condition members of one `parallel` or `select` group whose waits
+  are recorded next to each other, woken by one signal, no longer fail the next
+  task with `condition_wait_group_reopen_mismatch`, and each member consumes
+  its own reopen. A group whose condition waits are separated by another of its
+  members (for example a timer) is not covered yet. A wait that timed out
+  before the next open still ends it. Both histories were refused by 4.11.1 as
+  well.
+
 ## 4.11.2
 
 - A `wait_condition` timeout now runs from the first time the wait opens

src/cancellation_scope_replay.rs

@@ -626,9 +626,7 @@ impl WorkflowState {
     }
 
     fn next_scope_cancellation_sequence(&self) -> Result<u64> {
-        self.recorded_commands
-            .last()
-            .map_or(0, RecordedCommand::sequence)
+        self.max_recorded_sequence()
             .checked_add(self.commands.len() as u64)
             .and_then(|sequence| sequence.checked_add(1))
             .filter(|sequence| *sequence < i64::MAX as u64)
@@ -705,3 +703,34 @@ impl WorkflowContext {
             }))
     }
 }
+
+#[cfg(test)]
+mod reopen_order_tests {
+    use super::*;
+
+    // #193 review: the next authored sequence follows the highest recorded
+    // sequence even when grouped reopens are consumed out of sequence order.
+    #[test]
+    fn next_sequence_follows_the_highest_recorded_sequence() {
+        let mut state = WorkflowState::new(
+            crate::one_sided_grouped_reopen_history(Vec::new()),
+            "rust-workers".to_string(),
+            crate::DEFAULT_CODEC.to_string(),
+            None,
+        )
+        .expect("one-sided grouped reopen history");
+        let consumed = state
+            .recorded_commands
+            .iter()
+            .map(RecordedCommand::sequence)
+            .collect::<Vec<_>>();
+        assert_eq!(consumed, vec![1, 3, 2]);
+        state.command_cursor = state.recorded_commands.len();
+        assert_eq!(
+            state
+                .next_scope_cancellation_sequence()
+                .expect("next sequence"),
+            4
+        );
+    }
+}

src/cooperative_cancellation.rs

@@ -757,9 +757,7 @@ impl WorkflowState {
     fn next_cancellation_sequence(&self) -> Result<u64> {
         self.recorded_commands.get(self.command_cursor).map_or_else(
             || {
-                self.recorded_commands
-                    .last()
-                    .map_or(0, RecordedCommand::sequence)
+                self.max_recorded_sequence()
                     .checked_add(self.commands.len() as u64)
                     .and_then(|sequence| sequence.checked_add(1))
                     .filter(|sequence| *sequence < MAX_SEQUENCE)
@@ -2555,3 +2553,32 @@ impl WorkflowHandle {
             .await
     }
 }
+
+#[cfg(test)]
+mod reopen_order_tests {
+    use super::*;
+
+    // #193 review: the next authored sequence follows the highest recorded
+    // sequence even when grouped reopens are consumed out of sequence order.
+    #[test]
+    fn next_sequence_follows_the_highest_recorded_sequence() {
+        let mut state = WorkflowState::new(
+            crate::one_sided_grouped_reopen_history(Vec::new()),
+            "rust-workers".to_string(),
+            crate::DEFAULT_CODEC.to_string(),
+            None,
+        )
+        .expect("one-sided grouped reopen history");
+        let consumed = state
+            .recorded_commands
+            .iter()
+            .map(RecordedCommand::sequence)
+            .collect::<Vec<_>>();
+        assert_eq!(consumed, vec![1, 3, 2]);
+        state.command_cursor = state.recorded_commands.len();
+        assert_eq!(
+            state.next_cancellation_sequence().expect("next sequence"),
+            4
+        );
+    }
+}