From 7a906243954463c18f26603f07a3751f6be38995 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:10:34 -0600 Subject: [PATCH 1/2] Order every leaf of a group by member; pick scoped originals by sequence - group_condition_reopens_by_member now treats every grouped leaf of one group as part of the run: condition members are keyed by occurrence so their reopens join them, other leaves by sequence so they keep their place. A timer between condition members no longer strands their reopens (a1 t b1 a2 b2 -> a1 a2 t b1 b2). - A pending scope cancellation selects the group's original members by sequence and advances past the group's own commands, instead of slicing by position. - Moves the test helper and max_recorded_sequence below the reorder function so its doc comment is attached to it. Fixes #194 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 14 +- src/cancellation_scope_replay.rs | 89 +++++++++++- src/lib.rs | 242 ++++++++++++++++++++++++------- 3 files changed, 280 insertions(+), 65 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 089d9e1..b7755af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,13 +8,13 @@ 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. + Server 2.x, condition members of one `parallel` or `select` group woken by + one signal no longer fail the next task with + `condition_wait_group_reopen_mismatch`, and each member consumes its own + reopen, including when another member such as a timer sits between them + (#194). A pending scope cancellation over such a group selects its original + members by sequence. A wait that timed out before the next open still ends + it. Both histories were refused by 4.11.1 as well. - A `wait_condition` timeout now runs from the first time the wait opens (#187, parity with sdk-php#269). The Server wakes an open condition wait on any signal and records it satisfied. When the predicate was still false, the diff --git a/src/cancellation_scope_replay.rs b/src/cancellation_scope_replay.rs index ddcb89b..6af1d31 100644 --- a/src/cancellation_scope_replay.rs +++ b/src/cancellation_scope_replay.rs @@ -435,13 +435,10 @@ impl WorkflowState { if (sequence..sequence + span).all(resolved) { return Ok(false); } - let original: Vec<_> = self - .recorded_commands - .iter() - .skip(index) - .take(descriptors.len()) - .cloned() - .collect(); + // Recorded commands are in replay order, where a grouped condition + // member's reopens follow it (#193), so select the original members by + // sequence rather than by position. + let original = scoped_group_originals(&self.recorded_commands, sequence, span); self.validate_scoped_group_members(descriptors, &original)?; let boundary = CancellationDelivery { request_id: request.context.request_id().into(), @@ -468,7 +465,13 @@ impl WorkflowState { command_count: self.commands.len(), }); self.matched_recorded_pending = true; - self.command_cursor = index + descriptors.len(); + self.command_cursor = scoped_group_end( + &self.recorded_commands, + index, + sequence, + span, + &descriptors[0].group_path[0].parallel_group_id, + ); Ok(true) } @@ -704,6 +707,48 @@ impl WorkflowContext { } } +/// The group's original members, by sequence `base..base + span`. +fn scoped_group_originals( + commands: &[RecordedCommand], + base: u64, + span: u64, +) -> Vec { + (base..base.saturating_add(span)) + .filter_map(|sequence| { + commands + .iter() + .find(|command| command.sequence() == sequence) + .cloned() + }) + .collect() +} + +/// The cursor just past the group's commands starting at `index`: its original +/// members and any grouped condition reopens of the same group, which replay +/// order places next to them. +fn scoped_group_end( + commands: &[RecordedCommand], + index: usize, + base: u64, + span: u64, + group_id: &str, +) -> usize { + let owned = |command: &RecordedCommand| { + (base..base.saturating_add(span)).contains(&command.sequence()) + || matches!( + command, + RecordedCommand::ConditionWait { parallel_group_path: Some(path), .. } + if path.first().is_some_and(|entry| entry.parallel_group_id == group_id) + ) + }; + index + + commands + .iter() + .skip(index) + .take_while(|command| owned(command)) + .count() +} + #[cfg(test)] mod reopen_order_tests { use super::*; @@ -733,4 +778,32 @@ mod reopen_order_tests { 4 ); } + + // #194: with a one-sided reopen consumed as a1, a2, b1, a pending scope + // cancellation selects the originals a1 and b1 by sequence and moves past + // all three of the group's commands. + #[test] + fn scoped_group_selection_follows_sequence_not_position() { + let 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 originals = scoped_group_originals(&state.recorded_commands, 1, 2) + .iter() + .map(RecordedCommand::sequence) + .collect::>(); + assert_eq!(originals, vec![1, 2]); + assert_eq!( + scoped_group_end(&state.recorded_commands, 0, 1, 2, "parallel-calls:1:2"), + 3 + ); + // A command of another group ends the scoped group's commands. + assert_eq!( + scoped_group_end(&state.recorded_commands, 0, 1, 2, "parallel-calls:9:2"), + 1 + ); + } } diff --git a/src/lib.rs b/src/lib.rs index c2f1c83..d724b58 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17688,8 +17688,9 @@ fn recorded_commands( Ok(commands) } -/// Within a run of grouped condition waits from one parallel group, move each -/// physical reopen directly after the earlier waits of the same member. +/// Within a run of commands from one parallel group, move each physical +/// condition reopen directly after the earlier waits of the same member; other +/// leaves keep their place. /// /// Members replay in order from one shared cursor, and a member's chain reads /// only adjacent commands. When several members are woken in one task, the @@ -17697,6 +17698,73 @@ fn recorded_commands( /// chain would stop at its sibling and leave reopens unconsumed (#191). /// Outcomes are looked up by sequence and occurrence, so only consumption order /// changes. +fn group_condition_reopens_by_member(commands: Vec) -> Vec { + // Every leaf of a group belongs to the run; a condition member is keyed by + // its occurrence so its reopens join it, any other leaf by its sequence so + // it keeps its place (#194: a timer between condition members). + fn grouped_member(command: &RecordedCommand) -> Option<(String, String)> { + let (path, member) = match command { + RecordedCommand::ConditionWait { + occurrence_id, + parallel_group_path: Some(path), + .. + } => (path, format!("condition:{occurrence_id}")), + RecordedCommand::Activity { + parallel_group_path: Some(path), + .. + } + | RecordedCommand::Timer { + parallel_group_path: Some(path), + .. + } + | RecordedCommand::ChildWorkflow { + parallel_group_path: Some(path), + .. + } + | RecordedCommand::SignalWait { + parallel_group_path: Some(path), + .. + } => (path, format!("leaf:{}", command.sequence())), + _ => return None, + }; + path.first() + .map(|entry| (entry.parallel_group_id.clone(), member)) + } + let mut ordered = Vec::with_capacity(commands.len()); + let mut run: Vec = Vec::new(); + let mut run_group: Option = None; + let flush = |run: &mut Vec, ordered: &mut Vec| { + let mut members: Vec<(String, Vec)> = Vec::new(); + for command in run.drain(..) { + let occurrence = grouped_member(&command) + .map(|(_, occurrence)| occurrence) + .unwrap_or_default(); + match members.iter_mut().find(|(member, _)| *member == occurrence) { + Some((_, chain)) => chain.push(command), + None => members.push((occurrence, vec![command])), + } + } + ordered.extend(members.into_iter().flat_map(|(_, chain)| chain)); + }; + for command in commands { + match grouped_member(&command) { + Some((group, _)) if run_group.as_deref() == Some(group.as_str()) => run.push(command), + Some((group, _)) => { + flush(&mut run, &mut ordered); + run_group = Some(group); + run.push(command); + } + None => { + flush(&mut run, &mut ordered); + run_group = None; + ordered.push(command); + } + } + } + flush(&mut run, &mut ordered); + ordered +} + /// A two-member condition group where only member `a` was reopened: /// a1 (seq 1), b1 (seq 2), a1 satisfied, a2 (seq 3). Replay consumes it as /// a1, a2, b1, so the last recorded command is not the highest sequence. @@ -17743,54 +17811,6 @@ impl WorkflowState { } } -fn group_condition_reopens_by_member(commands: Vec) -> Vec { - fn grouped_member(command: &RecordedCommand) -> Option<(String, String)> { - match command { - RecordedCommand::ConditionWait { - occurrence_id, - parallel_group_path: Some(path), - .. - } => path - .first() - .map(|entry| (entry.parallel_group_id.clone(), occurrence_id.clone())), - _ => None, - } - } - let mut ordered = Vec::with_capacity(commands.len()); - let mut run: Vec = Vec::new(); - let mut run_group: Option = None; - let flush = |run: &mut Vec, ordered: &mut Vec| { - let mut members: Vec<(String, Vec)> = Vec::new(); - for command in run.drain(..) { - let occurrence = grouped_member(&command) - .map(|(_, occurrence)| occurrence) - .unwrap_or_default(); - match members.iter_mut().find(|(member, _)| *member == occurrence) { - Some((_, chain)) => chain.push(command), - None => members.push((occurrence, vec![command])), - } - } - ordered.extend(members.into_iter().flat_map(|(_, chain)| chain)); - }; - for command in commands { - match grouped_member(&command) { - Some((group, _)) if run_group.as_deref() == Some(group.as_str()) => run.push(command), - Some((group, _)) => { - flush(&mut run, &mut ordered); - run_group = Some(group); - run.push(command); - } - None => { - flush(&mut run, &mut ordered); - run_group = None; - ordered.push(command); - } - } - } - flush(&mut run, &mut ordered); - ordered -} - fn required_version_i32(payload: &Value, field: &str, sequence: u64) -> Result { payload .get(field) @@ -24709,6 +24729,25 @@ mod tests { vec![1, 2, 3, 4] ); + // A timer of the same group between condition members keeps its place + // while each condition member's reopens join it (#194): + // a1 t b1 a2 b2 -> a1 a2 t b1 b2. + assert_eq!( + order(vec![ + grouped_wait(1, 1, 3, 0), + RecordedCommand::Timer { + sequence: 2, + delay_seconds: 5, + fired: false, + parallel_group_path: Some(vec![parallel_group_entry(1, 3, 1, "condition")]), + }, + grouped_wait(3, 1, 3, 2), + grouped_wait(4, 1, 3, 0), + grouped_wait(5, 1, 3, 2), + ]), + vec![1, 4, 2, 3, 5] + ); + // Two different groups interleaved never merge into one run. assert_eq!( order(vec![ @@ -24807,6 +24846,109 @@ mod tests { } } + // #194: a timer member between two condition members. The Server reopens + // both condition members (votes 1, timer 2, approvals 3, votes 4, + // approvals 5); each member must still consume its own reopen. + #[test] + fn a_timer_between_grouped_condition_members_does_not_strand_their_reopens() { + let group = |index: u64| { + json!({ + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": index, + }) + }; + let member = |event_type: &str, sequence: u64, index: u64, key: &str, at: &str| { + let mut payload = group(index); + payload["sequence"] = json!(sequence); + payload["condition_wait_id"] = json!(format!("condition:{sequence}")); + payload["condition_wait_occurrence_id"] = json!(format!( + "rust:condition-wait:{}", + if index == 0 { 0 } else { 1 } + )); + payload["condition_key"] = json!(key); + payload["condition_definition_fingerprint"] = json!(format!("sha256:{key}-v1")); + payload["timeout_seconds"] = json!(300); + timed_event(event_type, at, payload) + }; + let timer = |event_type: &str, at: &str| { + let mut payload = group(1); + payload["sequence"] = json!(2); + payload["timer_id"] = json!("timer-2"); + payload["delay_seconds"] = json!(5); + timed_event(event_type, at, payload) + }; + let history = vec![ + member( + "ConditionWaitOpened", + 1, + 0, + "votes", + "2026-11-04T03:55:10.900000+00:00", + ), + timer("TimerScheduled", "2026-11-04T03:55:10.901000+00:00"), + member( + "ConditionWaitOpened", + 3, + 2, + "approvals", + "2026-11-04T03:55:10.902000+00:00", + ), + timer("TimerFired", "2026-11-04T03:55:15.901000+00:00"), + approval_signal("vote", "v1", "2026-11-04T03:55:16.000000+00:00"), + member( + "ConditionWaitSatisfied", + 1, + 0, + "votes", + "2026-11-04T03:55:16.100000+00:00", + ), + member( + "ConditionWaitOpened", + 4, + 0, + "votes", + "2026-11-04T03:55:16.101000+00:00", + ), + member( + "ConditionWaitOpened", + 5, + 2, + "approvals", + "2026-11-04T03:55:16.102000+00:00", + ), + approval_signal("vote", "v2", "2026-11-04T03:55:30.000000+00:00"), + approval_signal("approve", "a1", "2026-11-04T03:55:30.100000+00:00"), + ]; + let ctx = workflow_context(history); + let votes = ctx.clone(); + let approvals = ctx.clone(); + let mut group = Box::pin(ctx.parallel(vec![ + ParallelOperation::condition( + ConditionWaitOptions::new("votes", "sha256:votes-v1") + .timeout(Duration::from_secs(300)), + move || Ok(votes.signals("vote")?.len() >= 2), + ), + ParallelOperation::timer(Duration::from_secs(5)), + ParallelOperation::condition( + ConditionWaitOptions::new("approvals", "sha256:approvals-v1") + .timeout(Duration::from_secs(300)), + move || Ok(approvals.signals("approve")?.len() >= 1), + ), + ])); + let mut task_context = TaskContext::from_waker(noop_waker_ref()); + let poll = group.as_mut().poll(&mut task_context); + let Poll::Ready(Ok(results)) = poll else { + panic!("the group must complete: {poll:?}"); + }; + assert_eq!(results.len(), 3); + ctx.ensure_history_consumed() + .expect("every member consumes its own reopen"); + assert!(ctx.take_commands().expect("commands").is_empty()); + } + // #191: the PHP Server 2.11.8 history for two condition members woken by // one signal. It satisfies only the resume task's member (seq 2) but // records a reopen of the other, still-open member (seq 3 after seq 1). From cdc7a94c1b3438e3b5b6be231955874128b0b907 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 06:48:49 -0600 Subject: [PATCH 2/2] Cover timer-separated condition reopens in the official corpus --- regression-corpus-policy.json | 1 + ...parated-condition-group-server-reopen.json | 172 ++++++++++++++++++ tests/replay_regression_corpus.rs | 22 +++ 3 files changed, 195 insertions(+) create mode 100644 tests/fixtures/replay-regressions/timer-separated-condition-group-server-reopen.json diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 2852715..ad23d07 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -54,6 +54,7 @@ "History", "history", "fn continue_as_new_command\\b", + "fn group_condition_reopens_by_member\\b", "impl RecordedSnapshotValue", "impl ActivityRetrySnapshot", "fn recorded_optional_(?:u64|string)\\b", diff --git a/tests/fixtures/replay-regressions/timer-separated-condition-group-server-reopen.json b/tests/fixtures/replay-regressions/timer-separated-condition-group-server-reopen.json new file mode 100644 index 0000000..b2ba7ef --- /dev/null +++ b/tests/fixtures/replay-regressions/timer-separated-condition-group-server-reopen.json @@ -0,0 +1,172 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "rust-timer-separated-condition-group-server-reopen", + "protocol_version": "1.20", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.timer-separated-condition-group", + "payload_codec": "avro", + "input": [] + }, + "history": [ + { + "event_type": "ConditionWaitOpened", + "timestamp": "2026-11-04T03:55:10.900000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 0, + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "votes", + "condition_definition_fingerprint": "sha256:votes-v1", + "timeout_seconds": 300 + } + }, + { + "event_type": "TimerScheduled", + "timestamp": "2026-11-04T03:55:10.901000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 1, + "sequence": 2, + "timer_id": "timer-2", + "delay_seconds": 5 + } + }, + { + "event_type": "ConditionWaitOpened", + "timestamp": "2026-11-04T03:55:10.902000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 2, + "sequence": 3, + "condition_wait_id": "condition:3", + "condition_wait_occurrence_id": "rust:condition-wait:1", + "condition_key": "approvals", + "condition_definition_fingerprint": "sha256:approvals-v1", + "timeout_seconds": 300 + } + }, + { + "event_type": "TimerFired", + "timestamp": "2026-11-04T03:55:15.901000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 1, + "sequence": 2, + "timer_id": "timer-2", + "delay_seconds": 5 + } + }, + { + "event_type": "SignalReceived", + "timestamp": "2026-11-04T03:55:16.000000+00:00", + "payload": { + "signal_id": "v1", + "workflow_command_id": "command-v1", + "signal_name": "vote", + "arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgZBZGEA" + } + } + }, + { + "event_type": "ConditionWaitSatisfied", + "timestamp": "2026-11-04T03:55:16.100000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 0, + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "votes", + "condition_definition_fingerprint": "sha256:votes-v1", + "timeout_seconds": 300 + } + }, + { + "event_type": "ConditionWaitOpened", + "timestamp": "2026-11-04T03:55:16.101000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 0, + "sequence": 4, + "condition_wait_id": "condition:4", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "votes", + "condition_definition_fingerprint": "sha256:votes-v1", + "timeout_seconds": 300 + } + }, + { + "event_type": "ConditionWaitOpened", + "timestamp": "2026-11-04T03:55:16.102000+00:00", + "payload": { + "parallel_group_id": "parallel-calls:1:3", + "parallel_group_kind": "mixed", + "parallel_group_base_sequence": 1, + "parallel_group_size": 3, + "parallel_group_index": 2, + "sequence": 5, + "condition_wait_id": "condition:5", + "condition_wait_occurrence_id": "rust:condition-wait:1", + "condition_key": "approvals", + "condition_definition_fingerprint": "sha256:approvals-v1", + "timeout_seconds": 300 + } + }, + { + "event_type": "SignalReceived", + "timestamp": "2026-11-04T03:55:30.000000+00:00", + "payload": { + "signal_id": "v2", + "workflow_command_id": "command-v2", + "signal_name": "vote", + "arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgZBZGEA" + } + } + }, + { + "event_type": "SignalReceived", + "timestamp": "2026-11-04T03:55:30.100000+00:00", + "payload": { + "signal_id": "a1", + "workflow_command_id": "command-a1", + "signal_name": "approve", + "arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgZBZGEA" + } + } + } + ], + "expected": { + "type": "complete_workflow", + "result": "both" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 572927d..ea7a733 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -484,6 +484,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.grouped-condition-timed-reopen" | "corpus.selected-condition-timed-reopen" | "corpus.two-member-condition-group" + | "corpus.timer-separated-condition-group" | "corpus.cancellation-remaining-condition" | "corpus.durable-selection" | "corpus.durable-selection-portable-affinity" @@ -782,6 +783,27 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< Ok(json!(context.remaining()?.as_secs_f64())) }); } + "corpus.timer-separated-condition-group" => { + worker.register_workflow(workflow_type, |ctx, _| async move { + let votes = ctx.clone(); + let approvals = ctx.clone(); + ctx.parallel(vec![ + ParallelOperation::condition( + ConditionWaitOptions::new("votes", "sha256:votes-v1") + .timeout(Duration::from_secs(300)), + move || Ok(votes.signals("vote")?.len() >= 2), + ), + ParallelOperation::timer(Duration::from_secs(5)), + ParallelOperation::condition( + ConditionWaitOptions::new("approvals", "sha256:approvals-v1") + .timeout(Duration::from_secs(300)), + move || Ok(approvals.signals("approve")?.len() >= 1), + ), + ]) + .await?; + Ok(json!("both")) + }); + } "corpus.two-member-condition-group" => { worker.register_workflow(workflow_type, |ctx, _| async move { let votes = ctx.clone();