Add support and test multiple wait_for dependencies (#237)

go.mod

@@ -11,7 +11,7 @@ require (
 	go.flow.arcalot.io/dockerdeployer v0.7.3
 	go.flow.arcalot.io/expressions v0.4.5
 	go.flow.arcalot.io/kubernetesdeployer v0.10.0
-	go.flow.arcalot.io/pluginsdk v0.14.0
+	go.flow.arcalot.io/pluginsdk v0.14.1
 	go.flow.arcalot.io/podmandeployer v0.11.4
 	go.flow.arcalot.io/pythondeployer v0.6.2
 	go.flow.arcalot.io/testdeployer v0.6.2

go.sum

@@ -137,8 +137,8 @@ go.flow.arcalot.io/expressions v0.4.5 h1:GHRDHMkYIj2SN/TMc4aRApCewkJjl6moqhXjpdh
 go.flow.arcalot.io/expressions v0.4.5/go.mod h1:0Y2LgynO1SWA4bqsnKlCxqLME9zOR8tWKg3g+RG+FFQ=
 go.flow.arcalot.io/kubernetesdeployer v0.10.0 h1:4Fc6TsmM5pu1E0r9PzNu5AR1Q222B3Uo1Tqka1qqt8k=
 go.flow.arcalot.io/kubernetesdeployer v0.10.0/go.mod h1:OZTuKevUWxEHrlVYDE8+M5bHx1lJO6uwepKq+d3/k1w=
-go.flow.arcalot.io/pluginsdk v0.14.0 h1:GtMaVjhTLIDHfeqnUn2CCJM5BdJF7OVO+/CEcxArDrE=
-go.flow.arcalot.io/pluginsdk v0.14.0/go.mod h1:TOuJdxpyCcLYW+yNUBVe2vs6wFBaJj/z9+44IR1GqCU=
+go.flow.arcalot.io/pluginsdk v0.14.1 h1:S1PKJAXAvfFPmFXoyf7NrjNF+M4SaQzUC/1wWsmoEXU=
+go.flow.arcalot.io/pluginsdk v0.14.1/go.mod h1:TOuJdxpyCcLYW+yNUBVe2vs6wFBaJj/z9+44IR1GqCU=
 go.flow.arcalot.io/podmandeployer v0.11.4 h1:Oj0n1iW3X26dfTM2cgDUswtXNggVAmLpY5BUQqB8zBs=
 go.flow.arcalot.io/podmandeployer v0.11.4/go.mod h1:pc1gGXUAS8YeNkXLaVKApwmurqZ8VgrqZPakH8F/uUk=
 go.flow.arcalot.io/pythondeployer v0.6.2 h1:hIbDpdEhILArf8jMkqJplbxEnYYVcVslbH+IMdtXNy0=

workflow/workflow_test.go

@@ -411,6 +411,65 @@ func TestWaitForSerial(t *testing.T) {
 	}
 }
 
+var waitForMultipleWorkflowDefinition = `
+version: v0.2.0
+input:
+  root: RootObject
+  objects:
+    RootObject:
+      id: RootObject
+      properties: {}
+steps:
+  wait_a:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 5
+  wait_b:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 5
+  wait_c:
+    plugin:
+      src: "n/a"
+      deployment_type: "builtin"
+    step: wait
+    input:
+      wait_time_ms: 0
+    wait_for: # The map creates a dependency link for every property.
+      a: !expr $.steps.wait_a.outputs
+      b: !expr $.steps.wait_b.outputs
+outputs:
+  success:
+    wait_a: !soft-optional $.steps.wait_a.outputs
+    wait_b: !soft-optional $.steps.wait_b.outputs
+    wait_c: !expr $.steps.wait_c.outputs
+`
+
+func TestWaitForMultiple(t *testing.T) {
+	// For this test, a step waits for multiple independent steps with `wait_for`.
+	preparedWorkflow := assert.NoErrorR[workflow.ExecutableWorkflow](t)(
+		getTestImplPreparedWorkflow(t, waitForMultipleWorkflowDefinition),
+	)
+	outputID, outputData, err := preparedWorkflow.Execute(context.Background(), map[string]any{})
+	assert.NoError(t, err)
+	assert.Equals(t, outputID, "success")
+	// As an added check that the wait is actually occurring, C has no delay, but A and B do,
+	// and the output has a soft optional dependency on a and b, so if the values for A and B
+	// are present that means that C completed after A and B completed.
+	// This can be validated by commenting out the wait_for
+	// Soft optional does not enforce order, but wait_for does. This test should not be flaky
+	// as long as wait_for is working correctly.
+	typedOutputData := outputData.(map[interface{}]interface{})
+	assert.NotNil(t, typedOutputData["wait_a"])
+	assert.NotNil(t, typedOutputData["wait_b"])
+}
+
 var waitForStartedWorkflowDefinition = `
 version: v0.2.0
 input: