Support flat child-only all barriers

.github/workflows/server-parity.yml

@@ -28,6 +28,7 @@ on:
       - tests/Fixtures/HistoryBudget/**
       - tests/Fixtures/ServerUpgrade/**
       - tests/Fixtures/ServerParity/**
+      - tests/Fixtures/ServerParityObservations/**
       - tests/Fixtures/ServerParityBaselines/**
       - tests/Fixtures/ServerParityNormalization/**
       - tests/Unit/ServerParityRunnerTest.mjs
@@ -61,6 +62,7 @@ on:
       - tests/Unit/ServerParityOperatorMetricsContractTest.mjs
       - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs
       - tests/Unit/ServerParityConcurrentWaitsContractTest.mjs
+      - tests/Unit/ServerParityParallelChildAllContractTest.mjs
       - tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs
       - tests/Unit/ServerParityHistoryExportWaitsContractTest.mjs
       - tests/Unit/ServerParityContinueTransfersContractTest.mjs
@@ -2268,6 +2270,7 @@ jobs:
               tests/Unit/ServerParityOperatorMetricsContractTest.mjs \
               tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs \
               tests/Unit/ServerParityConcurrentWaitsContractTest.mjs \
+              tests/Unit/ServerParityParallelChildAllContractTest.mjs \
               tests/Unit/ServerParityClosedWaitTimeoutsContractTest.mjs \
               tests/Unit/ServerParityHistoryExportWaitsContractTest.mjs \
               tests/Unit/ServerParityContinueTransfersContractTest.mjs \

rust/src/runtime/all_groups.rs

@@ -1,4 +1,4 @@
-//! Published activity/timer all barriers retain every authored ancestor.
+//! Published all barriers retain every qualified authored ancestor.
 use super::{Result, refuse};
 use axum::http::StatusCode;
 use serde_json::{Value, json};
@@ -38,6 +38,7 @@ fn entry(value: &Value, leaf: &str) -> Result<Value> {
     let prefix = match kind {
         "activity" => "parallel-activities",
         "timer" => "parallel-timers",
+        "child" => "parallel-children",
         _ => "parallel-calls",
     };
     if value["parallel_group_id"] != format!("{prefix}:{base}:{size}") {
@@ -50,14 +51,19 @@ fn entry(value: &Value, leaf: &str) -> Result<Value> {
 }
 
 pub(super) fn metadata(value: &Value, kind: &str) -> Result<Value> {
-    if !matches!(kind, "activity" | "timer") {
+    if !matches!(kind, "activity" | "timer" | "child") {
         return Err(invalid());
     }
     let top = entry(value, kind)?;
     let path = value["parallel_group_path"]
         .as_array()
         .filter(|path| !path.is_empty())
         .ok_or_else(invalid)?;
+    // Child selections, nested barriers and mixed child groups have not been
+    // qualified. Keep their refusal independent of activity/timer paths.
+    if kind == "child" && (path.len() != 1 || top["parallel_group_kind"] != "child") {
+        return Err(invalid());
+    }
     let path = path
         .iter()
         .map(|value| entry(value, kind))
@@ -95,6 +101,7 @@ pub(super) fn validate_batch(commands: &[Value]) -> Result<()> {
         let kind = match command["type"].as_str() {
             Some("schedule_activity") => "activity",
             Some("start_timer") => "timer",
+            Some("start_child_workflow") => "child",
             _ => return Err(invalid()),
         };
         let metadata = metadata(command, kind)?;
@@ -156,6 +163,8 @@ impl AllHistory {
             ("activity", "activity_execution_id")
         } else if event.starts_with("Timer") {
             ("timer", "timer_id")
+        } else if event.starts_with("Child") {
+            ("child", "child_call_id")
         } else {
             return true;
         };
@@ -196,7 +205,10 @@ impl AllHistory {
         {
             return false;
         }
-        if matches!(event, "ActivityScheduled" | "TimerScheduled") {
+        if matches!(
+            event,
+            "ActivityScheduled" | "TimerScheduled" | "ChildWorkflowScheduled"
+        ) {
             let group = self.groups.entry(group_id).or_insert_with(|| Group {
                 size,
                 members: BTreeMap::new(),
@@ -243,6 +255,10 @@ impl AllHistory {
                 | "ActivityCancelled"
                 | "TimerFired"
                 | "TimerCancelled"
+                | "ChildRunCompleted"
+                | "ChildRunFailed"
+                | "ChildRunCancelled"
+                | "ChildRunTerminated"
         ) {
             member.terminal = true;
             if group.members.len() == size && group.members.values().all(|member| member.terminal) {

rust/src/runtime/child_summary.rs

@@ -18,6 +18,7 @@ pub(super) struct ChildSummaryHistory {
 
 struct ExpectedChild {
     id: String,
+    parallel: Value,
     /// The first run, which the parent's wait keeps naming.
     run: String,
     instance: String,
@@ -58,7 +59,12 @@ impl ChildSummaryHistory {
         if !previous.same_call(payload) || previous.status != "waiting" {
             return false;
         }
-        let text = |key: &str| payload[key].as_str().filter(|v| !v.is_empty()).map(str::to_owned);
+        let text = |key: &str| {
+            payload[key]
+                .as_str()
+                .filter(|v| !v.is_empty())
+                .map(str::to_owned)
+        };
         match event {
             "ChildRunStarted" if !previous.started => {
                 if payload["child_run_number"] != 1
@@ -69,10 +75,15 @@ impl ChildSummaryHistory {
                 }
                 previous.started = true;
             }
-            "ChildRunStarted" if payload["retry_attempt"].is_i64() => {
-                let (Some(run), Some(link), Some(number)) =
-                    (text("child_workflow_run_id"), text("workflow_link_id"), payload["child_run_number"].as_i64())
-                else {
+            "ChildRunStarted"
+                if payload["retry_attempt"].is_i64()
+                    || previous.parallel.get("parallel_group_id").is_some() =>
+            {
+                let (Some(run), Some(link), Some(number)) = (
+                    text("child_workflow_run_id"),
+                    text("workflow_link_id"),
+                    payload["child_run_number"].as_i64(),
+                ) else {
                     return false;
                 };
                 if number <= previous.run_number || run == previous.latest_run || link == previous.latest_link {
@@ -116,15 +127,14 @@ impl ExpectedChild {
         };
         let id = text("child_call_id")?;
         let label = text("child_workflow_type")?;
-        if payload["workflow_link_id"] != id
-            || payload["child_workflow_class"] != label
-            || !payload["parallel_group_id"].is_null()
-            || !payload["parallel_group_path"].is_null()
-        {
+        let parallel = super::selections::metadata(payload, "child").ok()?;
+        super::selections::check_sequence(&parallel, payload["sequence"].as_i64()?).ok()?;
+        if payload["workflow_link_id"] != id || payload["child_workflow_class"] != label {
             return None;
         }
         Some(Self {
             id,
+            parallel,
             run: text("child_workflow_run_id")?,
             instance: text("child_workflow_instance_id")?,
             label,
@@ -142,14 +152,13 @@ impl ExpectedChild {
             && payload["child_workflow_instance_id"] == self.instance.as_str()
             && payload["child_workflow_type"] == self.label.as_str()
             && payload["child_workflow_class"] == self.label.as_str()
-            && payload["parallel_group_id"].is_null()
-            && payload["parallel_group_path"].is_null()
+            && super::selections::metadata(payload, "child").ok().as_ref() == Some(&self.parallel)
     }
 
     fn wait_status(&self) -> &'static str {
         match self.status {
             "waiting" => "open",
-            "cancelled" => "cancelled",
+            "cancelled" | "terminated" => "cancelled",
             _ => "resolved",
         }
     }
@@ -173,7 +182,7 @@ where
         history: &ChildSummaryHistory,
         summary: &mut ActiveSummary,
     ) -> Result<bool> {
-        let rows = Self::query("SELECT w.*,c.status AS call_status,c.resolved_child_run_id,c.resolved_child_instance_id,c.child_workflow_type,c.metadata AS call_metadata,l.id AS link_id,l.child_workflow_run_id AS link_run,l.child_workflow_instance_id AS link_instance FROM workflow_run_waits w LEFT JOIN workflow_child_calls c ON c.parent_workflow_run_id=w.workflow_run_id AND c.sequence=w.sequence LEFT JOIN workflow_links l ON l.parent_workflow_run_id=w.workflow_run_id AND l.sequence=w.sequence AND l.link_type='child_workflow' AND l.child_workflow_run_id=c.resolved_child_run_id WHERE w.workflow_run_id=$1 AND w.kind='child' ORDER BY w.sequence")
+        let rows = Self::query("SELECT w.*,c.status AS call_status,c.resolved_child_run_id,c.resolved_child_instance_id,c.child_workflow_type,c.metadata AS call_metadata,l.id AS link_id,l.child_workflow_run_id AS link_run,l.child_workflow_instance_id AS link_instance,l.parallel_group_path AS link_group_path FROM workflow_run_waits w LEFT JOIN workflow_child_calls c ON c.parent_workflow_run_id=w.workflow_run_id AND c.sequence=w.sequence LEFT JOIN workflow_links l ON l.parent_workflow_run_id=w.workflow_run_id AND l.sequence=w.sequence AND l.link_type='child_workflow' AND l.child_workflow_run_id=c.resolved_child_run_id WHERE w.workflow_run_id=$1 AND w.kind='child' ORDER BY w.sequence")
             .bind(run_id).fetch_all(&mut **tx).await?;
         if rows.len() != history.calls.len() {
             return Ok(false);
@@ -232,7 +241,7 @@ where
         };
         let pending = rows.iter().find(|row| {
             DB::string(row, "source_status").is_ok_and(|status| {
-                matches!(status.as_str(), "cancelled" | "failed")
+                matches!(status.as_str(), "cancelled" | "terminated" | "failed")
                     || (status == "completed" && open.is_none())
             })
         });
@@ -307,6 +316,8 @@ where
                 .is_some_and(|v| v["child_call_id"] == child.id)
             && matches("link_id", &child.latest_link)?
             && matches("link_run", &child.latest_run)?
-            && matches("link_instance", &child.instance)?)
+            && matches("link_instance", &child.instance)?
+            && DB::optional_document_row(row, "link_group_path")?
+                == child.parallel.get("parallel_group_path").cloned())
     }
 }