Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 37 additions & 26 deletions workflow/builder.go
Original file line number Diff line number Diff line change
Expand Up @@ -227,11 +227,23 @@ func (wb *Builder) AddFanInBarrierEdge(sources []ExecutorBinding, target Executo
return wb
}

// BuildOptions configures workflow validation during [Builder.BuildWithOptions].
type BuildOptions struct {
DisableOrphanValidation bool
}
Comment thread
qmuntal marked this conversation as resolved.

// Build validates the assembled graph, including orphan-executor checks, and
// returns the immutable *Workflow. It returns the first accumulated error if
// returns the immutable [Workflow]. It returns the first accumulated error if
// any builder step failed or validation did not pass.
func (wb *Builder) Build() (*Workflow, error) {
return wb.build(true)
return wb.BuildWithOptions(BuildOptions{})
}

// BuildWithOptions validates the assembled graph according to opts and returns
// the immutable [Workflow]. Orphan-executor checks can be disabled; all other
// validation is always performed.
func (wb *Builder) BuildWithOptions(opts BuildOptions) (*Workflow, error) {
return wb.build(!opts.DisableOrphanValidation)
}

func (wb *Builder) build(validateOrphans bool) (*Workflow, error) {
Expand Down Expand Up @@ -296,33 +308,32 @@ func (wb *Builder) validate(validateOrphans bool) bool {
return false
}
}
if !validateOrphans || len(wb.executorsBindings) == 0 {
return true
}
// Make sure that all nodes are connected to the start executor (transitively).
remainingExecutors := make(map[string]struct{}, len(wb.executorsBindings))
for id := range wb.executorsBindings {
remainingExecutors[id] = struct{}{}
}
toVisit := []string{wb.startExecutorId}
for len(toVisit) > 0 {
var currentID string
currentID, toVisit = toVisit[0], toVisit[1:]
if _, unvisited := remainingExecutors[currentID]; !unvisited {
continue
if validateOrphans && len(wb.executorsBindings) > 0 {
// Make sure that all nodes are connected to the start executor (transitively).
remainingExecutors := make(map[string]struct{}, len(wb.executorsBindings))
for id := range wb.executorsBindings {
remainingExecutors[id] = struct{}{}
}
delete(remainingExecutors, currentID)
if edges, ok := wb.edges[currentID]; ok {
for _, edge := range edges {
toVisit = append(toVisit, edge.Connection.SinkIDs...)
toVisit := []string{wb.startExecutorId}
for len(toVisit) > 0 {
var currentID string
currentID, toVisit = toVisit[0], toVisit[1:]
if _, unvisited := remainingExecutors[currentID]; !unvisited {
continue
}
delete(remainingExecutors, currentID)
if edges, ok := wb.edges[currentID]; ok {
for _, edge := range edges {
toVisit = append(toVisit, edge.Connection.SinkIDs...)
}
}
}
}
if len(remainingExecutors) > 0 {
keys := slices.Collect(maps.Keys(remainingExecutors))
slices.Sort(keys)
wb.err = fmt.Errorf("workflow cannot be built because there are orphaned executors: %v", keys)
return false
if len(remainingExecutors) > 0 {
keys := slices.Collect(maps.Keys(remainingExecutors))
slices.Sort(keys)
wb.err = fmt.Errorf("workflow cannot be built because there are orphaned executors: %v", keys)
return false
}
}
// Warn about self-loops (executor connecting to itself), which may cause infinite
// recursion if not gated by a condition.
Expand Down
35 changes: 35 additions & 0 deletions workflow/builder_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -202,6 +202,41 @@ func TestBuilder_Validation_FailsWhenUnreachableExecutors(t *testing.T) {
}
}

func TestBuilder_BuildWithOptionsCanSkipOrphanValidation(t *testing.T) {
wf, err := workflow.NewBuilder(newPlaceholder("start")).
BindExecutor(newNoOpExecutor("start")).
AddEdge(newNoOpExecutor("unreachable"), newNoOpExecutor("also-unreachable")).
BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true})
if err != nil {
t.Fatalf("BuildWithOptions() error = %v, want nil", err)
}
if wf == nil {
t.Fatal("BuildWithOptions() workflow = nil, want workflow")
}
}

func TestBuilder_BuildWithOptionsValidatesOrphansByDefault(t *testing.T) {
_, err := workflow.NewBuilder(newPlaceholder("start")).
BindExecutor(newNoOpExecutor("start")).
AddEdge(newNoOpExecutor("unreachable"), newNoOpExecutor("also-unreachable")).
BuildWithOptions(workflow.BuildOptions{})
if err == nil || !strings.Contains(err.Error(), "orphaned executors") {
t.Fatalf("BuildWithOptions() error = %v, want orphaned executors error", err)
}
}

func TestBuilder_BuildWithOptionsStillValidatesTypeCompatibility(t *testing.T) {
source := newTypedExecutor[string, int]("unreachable")
target := newTypedExecutor[string, string]("also-unreachable")

_, err := workflow.NewBuilder(newNoOpExecutor("start")).
AddEdge(source, target).
BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true})
if err == nil || !strings.Contains(err.Error(), "type incompatibility") {
t.Fatalf("BuildWithOptions() error = %v, want type incompatibility error", err)
}
}

func TestBuilder_Validation_AddEdgesOutOfOrderDoesNotImpactReachability(t *testing.T) {
wf, err := workflow.NewBuilder(newPlaceholder("start")).
BindExecutor(newNoOpExecutor("start")).
Expand Down
Loading