pkg/workflows/sdk: add WorkflowSpecFactory.BeginSerial/BeginAsync

Makefile

@@ -47,3 +47,7 @@ lint-workspace:
 
 lint:
 	@./script/lint.sh $(GOLANGCI_LINT_VERSION) "$(GOLANGCI_LINT_COMMON_OPTS)" $(GOLANGCI_LINT_DIRECTORY) "--new-from-rev=origin/main"
+
+.PHONY: test-quiet
+test-quiet:
+	go test ./... | grep -v "\[no test files\]" | grep -v "\(cached\)"

go.mod

@@ -1,8 +1,6 @@
 module github.com/smartcontractkit/chainlink-common
 
-go 1.22.0
-
-toolchain go1.22.7
+go 1.23
 
 require (
 	github.com/andybalholm/brotli v1.1.0

pkg/capabilities/capabilities.go

@@ -1,6 +1,7 @@
 package capabilities
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"regexp"
@@ -53,6 +54,26 @@ func (c CapabilityType) IsValid() error {
 	return fmt.Errorf("invalid capability type: %s", c)
 }
 
+func (c CapabilityType) cmpOrder() int {
+	switch c {
+	case CapabilityTypeTrigger:
+		return 0
+	case CapabilityTypeAction:
+		return 1
+	case CapabilityTypeConsensus:
+		return 2
+	case CapabilityTypeTarget:
+		return 3
+	case CapabilityTypeUnknown:
+		return 4
+	default:
+		return 5
+	}
+}
+func (c CapabilityType) Compare(c2 CapabilityType) int {
+	return cmp.Compare(c.cmpOrder(), c2.cmpOrder())
+}
+
 // CapabilityResponse is a struct for the Execute response of a capability.
 type CapabilityResponse struct {
 	Value *values.Map

pkg/workflows/models_yaml_test.go

@@ -49,7 +49,7 @@ var transformJSON = cmp.FilterValues(func(x, y []byte) bool {
 	return out
 }))
 
-func TestWorkflowSpecMarshalling(t *testing.T) {
+func TestWorkflowSpecYamlMarshalling(t *testing.T) {
 	t.Parallel()
 	fixtureReader := yamlFixtureReaderBytes(t, "marshalling")
 

pkg/workflows/sdk/builder.go

@@ -15,12 +15,22 @@ type WorkflowSpecFactory struct {
 	emptyNames     bool
 	badCapTypes    []string
 	fns            map[string]func(runtime Runtime, request capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error)
+	serialMode     bool
+	prevRefs       []string
 }
 
 func (w *WorkflowSpecFactory) GetFn(name string) func(sdk Runtime, request capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) {
 	return w.fns[name]
 }
 
+func (w *WorkflowSpecFactory) BeginSerial() {
+	w.serialMode = true
+}
+
+func (w *WorkflowSpecFactory) BeginAsync() {
+	w.serialMode = false
+}
+
 type CapDefinition[O any] interface {
 	capDefinition
 	self() CapDefinition[O]
@@ -107,6 +117,15 @@ type NewWorkflowParams struct {
 	Name  string
 }
 
+// NewSerialWorkflowSpecFactory returns a new WorkflowSpecFactory in Serial mode.
+// This is the same as calling NewWorkflowSpecFactory then WorkflowSpecFactory.BeginSerial.
+func NewSerialWorkflowSpecFactory(params NewWorkflowParams) *WorkflowSpecFactory {
+	f := NewWorkflowSpecFactory(params)
+	f.BeginSerial()
+	return f
+}
+
+// NewWorkflowSpecFactory returns a new NewWorkflowSpecFactory.
 func NewWorkflowSpecFactory(
 	params NewWorkflowParams,
 ) *WorkflowSpecFactory {
@@ -128,6 +147,16 @@ func NewWorkflowSpecFactory(
 // AddTo is meant to be called by generated code
 func (step *Step[O]) AddTo(w *WorkflowSpecFactory) CapDefinition[O] {
 	stepDefinition := step.Definition
+
+	if w.serialMode {
+		// ensure we depend on each previous step
+		for _, prevRef := range w.prevRefs {
+			if !stepDefinition.Inputs.HasRef(prevRef) {
+				stepDefinition.Condition = fmt.Sprintf("$(%s.success)", prevRef)
+			}
+		}
+	}
+
 	stepRef := stepDefinition.Ref
 	if w.names[stepRef] && stepDefinition.CapabilityType != capabilities.CapabilityTypeTarget {
 		w.duplicateNames[stepRef] = true
@@ -152,7 +181,14 @@ func (step *Step[O]) AddTo(w *WorkflowSpecFactory) CapDefinition[O] {
 		w.badCapTypes = append(w.badCapTypes, stepDefinition.ID)
 	}
 
-	return &capDefinitionImpl[O]{ref: fmt.Sprintf("$(%s.outputs)", step.Definition.Ref)}
+	c := &capDefinitionImpl[O]{ref: fmt.Sprintf("$(%s.outputs)", step.Definition.Ref)}
+
+	if w.serialMode {
+		w.prevRefs = []string{step.Definition.Ref}
+	} else {
+		w.prevRefs = append(w.prevRefs, step.Definition.Ref)
+	}
+	return c
 }
 
 // AccessField is meant to be used by generated code

pkg/workflows/sdk/builder_test.go

@@ -7,7 +7,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"sigs.k8s.io/yaml"
 
-	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	ocr3 "github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/ocr3cap"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/targets/chainwriter"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/triggers/streams"
@@ -205,81 +204,7 @@ func TestBuilder_ValidSpec(t *testing.T) {
 		actual, err := factory.Spec()
 		require.NoError(t, err)
 
-		expected := sdk.WorkflowSpec{
-			Name:  "notccipethsep",
-			Owner: "0x00000000000000000000000000000000000000aa",
-			Triggers: []sdk.StepDefinition{
-				{
-					ID:             "[email protected]",
-					Ref:            "trigger",
-					Inputs:         sdk.StepInputs{},
-					Config:         map[string]any{"maxFrequencyMs": 5000},
-					CapabilityType: capabilities.CapabilityTypeTrigger,
-				},
-			},
-			Actions: make([]sdk.StepDefinition, 0),
-			Consensus: []sdk.StepDefinition{
-				{
-					ID:  "[email protected]",
-					Ref: "data-feeds-report",
-					Inputs: sdk.StepInputs{
-						Mapping: map[string]any{"observations": []map[string]any{
-							{
-								"Metadata": map[string]any{
-									"MinRequiredSignatures": 1,
-									"Signers":               []string{"$(trigger.outputs.Metadata.Signer)"},
-								},
-								"Payload": []map[string]any{
-									{
-										"BenchmarkPrice":       "$(trigger.outputs.Payload.BuyPrice)",
-										"FeedID":               anyFakeFeedID,
-										"FullReport":           "$(trigger.outputs.Payload.FullReport)",
-										"ObservationTimestamp": "$(trigger.outputs.Payload.ObservationTimestamp)",
-										"ReportContext":        "$(trigger.outputs.Payload.ReportContext)",
-										"Signatures":           []string{"$(trigger.outputs.Payload.Signature)"},
-									},
-								},
-								"Timestamp": "$(trigger.outputs.Timestamp)",
-							},
-						}},
-					},
-					Config: map[string]any{
-						"aggregation_config": ocr3.DataFeedsConsensusConfigAggregationConfig{
-							AllowedPartialStaleness: "0.5",
-							Feeds: map[string]ocr3.FeedValue{
-								anyFakeFeedID: {
-									Deviation: "0.5",
-									Heartbeat: 3600,
-								},
-							},
-						},
-						"aggregation_method": "data_feeds",
-						"encoder":            "EVM",
-						"encoder_config": ocr3.EncoderConfig{
-							"Abi": "(bytes32 FeedID, uint224 Price, uint32 Timestamp)[] Reports",
-						},
-						"report_id": "0001",
-					},
-					CapabilityType: capabilities.CapabilityTypeConsensus,
-				},
-			},
-			Targets: []sdk.StepDefinition{
-				{
-					ID: "[email protected]",
-					Inputs: sdk.StepInputs{
-						Mapping: map[string]any{"signed_report": "$(data-feeds-report.outputs)"},
-					},
-					Config: map[string]any{
-						"address":    "0xE0082363396985ae2FdcC3a9F816A586Eed88416",
-						"deltaStage": "45s",
-						"schedule":   "oneAtATime",
-					},
-					CapabilityType: capabilities.CapabilityTypeTarget,
-				},
-			},
-		}
-
-		testutils.AssertWorkflowSpec(t, expected, actual)
+		testutils.AssertWorkflowSpec(t, notStreamSepoliaWorkflowSpec, actual)
 	})
 
 	t.Run("duplicate names causes errors", func(t *testing.T) {

pkg/workflows/sdk/compute_test.go

@@ -39,74 +39,8 @@ func TestCompute(t *testing.T) {
 
 		spec, err2 := workflow.Spec()
 		require.NoError(t, err2)
-		expectedSpec := sdk.WorkflowSpec{
-			Name:  "name",
-			Owner: "owner",
-			Triggers: []sdk.StepDefinition{
-				{
-					ID:             "[email protected]",
-					Ref:            "trigger",
-					Inputs:         sdk.StepInputs{},
-					Config:         map[string]any{"maxFrequencyMs": 5000},
-					CapabilityType: capabilities.CapabilityTypeTrigger,
-				},
-			},
-			Actions: []sdk.StepDefinition{
-				{
-					ID:  "[email protected]",
-					Ref: "Compute",
-					Inputs: sdk.StepInputs{
-						Mapping: map[string]any{"Arg0": "$(trigger.outputs)"},
-					},
-					Config: map[string]any{
-						"binary": "$(ENV.binary)",
-						"config": "$(ENV.config)",
-					},
-					CapabilityType: capabilities.CapabilityTypeAction,
-				},
-			},
-			Consensus: []sdk.StepDefinition{
-				{
-					ID:  "[email protected]",
-					Ref: "data-feeds-report",
-					Inputs: sdk.StepInputs{
-						Mapping: map[string]any{"observations": "$(Compute.outputs.Value)"},
-					},
-					Config: map[string]any{
-						"aggregation_config": ocr3.DataFeedsConsensusConfigAggregationConfig{
-							AllowedPartialStaleness: "false",
-							Feeds: map[string]ocr3.FeedValue{
-								anyFakeFeedID: {
-									Deviation: "0.5",
-									Heartbeat: 3600,
-								},
-							},
-						},
-						"aggregation_method": "data_feeds",
-						"encoder":            ocr3.EncoderEVM,
-						"encoder_config":     ocr3.EncoderConfig{},
-						"report_id":          "0001",
-					},
-					CapabilityType: capabilities.CapabilityTypeConsensus,
-				},
-			},
-			Targets: []sdk.StepDefinition{
-				{
-					ID: "[email protected]",
-					Inputs: sdk.StepInputs{
-						Mapping: map[string]any{"signed_report": "$(data-feeds-report.outputs)"},
-					},
-					Config: map[string]any{
-						"address":    "0xE0082363396985ae2FdcC3a9F816A586Eed88416",
-						"deltaStage": "45s",
-						"schedule":   "oneAtATime",
-					},
-					CapabilityType: capabilities.CapabilityTypeTarget,
-				},
-			},
-		}
 
-		testutils.AssertWorkflowSpec(t, expectedSpec, spec)
+		testutils.AssertWorkflowSpec(t, serialWorkflowSpec, spec)
 	})
 
 	t.Run("compute runs the function and returns the value", func(t *testing.T) {
@@ -133,7 +67,7 @@ func TestCompute(t *testing.T) {
 func createWorkflow(fn func(_ sdk.Runtime, inputFeed notstreams.Feed) ([]streams.Feed, error)) *sdk.WorkflowSpecFactory {
 	workflow := sdk.NewWorkflowSpecFactory(sdk.NewWorkflowParams{
 		Owner: "owner",
-		Name:  "name",
+		Name:  "serial",
 	})
 
 	trigger := notstreams.TriggerConfig{MaxFrequencyMs: 5000}.New(workflow)

pkg/workflows/sdk/helper_test.go

@@ -0,0 +1,14 @@
+package sdk
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func (w *WorkflowSpecFactory) MustSpec(t *testing.T) WorkflowSpec {
+	t.Helper()
+	s, err := w.Spec()
+	require.NoError(t, err)
+	return s
+}

pkg/workflows/sdk/testdata/fixtures/charts/README.md

@@ -0,0 +1,8 @@
+# WorkflowSpec Charts
+
+This directory contains WorkflowSpec chart golden files. They are validated against test data by TestWorkflowSpecFormatChart, 
+and can be regenerated by passing the `-update` flag:
+```sh
+go test -run=TestWorkflowSpecFormatChart ./pkg/workflows/sdk/ -update
+```
+You can also invoke go:generate on package sdk, which will do the same.

pkg/workflows/sdk/testdata/fixtures/charts/builder_parallel.md

@@ -0,0 +1,26 @@
+```mermaid
+flowchart
+
+	trigger[\"<b>trigger</b><br>trigger<br><i>(basic-test-trigger[at]1.0.0)</i>"/]
+
+	compute["<b>compute</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	get-bar -- Value --> compute 
+			get-baz -- Value --> compute 
+			get-foo -- Value --> compute 
+
+	get-bar["<b>get-bar</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	trigger -- cool_output --> get-bar 
+
+	get-baz["<b>get-baz</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	trigger -- cool_output --> get-baz 
+
+	get-foo["<b>get-foo</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	trigger -- cool_output --> get-foo 
+
+	consensus[["<b>consensus</b><br>consensus<br><i>(offchain_reporting[at]1.0.0)</i>"]]
+	compute -- Value --> consensus 
+
+	unnamed6[/"target<br><i>(id)</i>"\]
+	consensus --> unnamed6
+
+```

pkg/workflows/sdk/testdata/fixtures/charts/builder_serial.md

@@ -0,0 +1,28 @@
+```mermaid
+flowchart
+
+	trigger[\"<b>trigger</b><br>trigger<br><i>(basic-test-trigger[at]1.0.0)</i>"/]
+
+	compute["<b>compute</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	get-bar -- Value --> compute 
+			get-baz -- Value --> compute 
+			get-foo -- Value --> compute 
+
+	get-bar["<b>get-bar</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	get-foo -..-> get-bar
+	trigger -- cool_output --> get-bar 
+
+	get-baz["<b>get-baz</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	get-bar -..-> get-baz
+	trigger -- cool_output --> get-baz 
+
+	get-foo["<b>get-foo</b><br>action<br><i>(custom_compute[at]1.0.0)</i>"]
+	trigger -- cool_output --> get-foo 
+
+	consensus[["<b>consensus</b><br>consensus<br><i>(offchain_reporting[at]1.0.0)</i>"]]
+	compute -- Value --> consensus 
+
+	unnamed6[/"target<br><i>(id)</i>"\]
+	consensus --> unnamed6
+
+```