Move _disable_concurrency to static field to avoid unnecessary input lookups

Author: sebastianrath

core/base.go

@@ -169,6 +169,9 @@ type NodeBaseInterface interface {
 	IsExecutionNode() bool
 	SetExecutionNode(execNode bool)
 
+	DisableConcurrency() bool
+	SetDisableConcurrency(v bool)
+
 	// Returns the cache type where data is stored or should be stored to
 	// By default this depends on if this is an execution node or not.
 	GetCacheType() CacheType
@@ -183,9 +186,10 @@ type NodeBaseComponent struct {
 	FullPath        string // Full path of the node within the graph hierarchy
 	CacheId         string // Unique identifier for the cache
 	NodeType        string // Node type of the node (e.g. core/run@v1 or github.com/actions/checkout@v3)
-	Graph           *ActionGraph
-	Parent          NodeBaseInterface
-	isExecutionNode bool
+	Graph              *ActionGraph
+	Parent             NodeBaseInterface
+	isExecutionNode    bool
+	disableConcurrency bool
 }
 
 func (n *NodeBaseComponent) GetCacheType() CacheType {
@@ -204,6 +208,14 @@ func (n *NodeBaseComponent) SetExecutionNode(execNode bool) {
 	n.isExecutionNode = execNode
 }
 
+func (n *NodeBaseComponent) DisableConcurrency() bool {
+	return n.disableConcurrency
+}
+
+func (n *NodeBaseComponent) SetDisableConcurrency(v bool) {
+	n.disableConcurrency = v
+}
+
 func (n *NodeBaseComponent) SetId(id string) {
 	n.Id = id
 	n.CacheId = fmt.Sprintf("%s:%s", n.Id, uuid.New().String())
@@ -562,21 +574,6 @@ func RegisterNodeFactory(nodeDefStr string, fn nodeFactoryFunc) error {
 		nodeDef.Outputs[outputId] = outputDef
 	}
 
-	// inject the hidden _disable_concurrency input for execution nodes.
-	// When set to true at runtime, concurrent calls to this nodes
-	// ExecuteImpl are serialized via a per-node-ID mutex
-	if countExec > 0 {
-		nodeDef.Inputs[InputId("_disable_concurrency")] = InputDefinition{
-			PortDefinition: PortDefinition{
-				Name:  "Disable Concurrency",
-				Type:  "bool",
-				Index: 999999,
-			},
-			HideSocket: true,
-			Default:    false,
-		}
-	}
-
 	id := fmt.Sprintf("%v@v%v", nodeDef.Id, nodeDef.Version)
 	_, ok := registries[id]
 	if ok {

core/executions.go

@@ -79,28 +79,24 @@ func (n *Executions) Execute(outputPort OutputId, ec *ExecutionState, err error)
 		return nil
 	}
 
-	// If the destination node has _disable_concurrency set to true, serialize execution
+	// If the destination node has _disable_concurrency set, serialize execution
 	// through a per-node-ID mutex to prevent concurrent ExecuteImpl calls.
 	// The lock is stored as pending and released when the node calls Execute
 	// to dispatch downstream (above), or as a fallback when ExecuteImpl
 	// returns without dispatching (below).
-	var dcNodeId string
-	if nwi, ok := dest.DstNode.(NodeWithInputs); ok {
-		dcVal, dcErr := InputValueById[bool](ec, nwi, InputId("_disable_concurrency"))
-		if dcErr == nil && dcVal {
-			dcNodeId = dest.DstNode.GetId()
-			actual, _ := ec.Graph.ConcurrencyLocks.LoadOrStore(dcNodeId, &sync.Mutex{})
-			mu := actual.(*sync.Mutex)
-			mu.Lock()
-			ec.PendingConcurrencyLocks.Store(dcNodeId, mu)
-			// Fallback: release if ExecuteImpl returns without calling Execute
-			// (end of chain, error, or panic).
-			defer func() {
-				if lockVal, loaded := ec.PendingConcurrencyLocks.LoadAndDelete(dcNodeId); loaded {
-					lockVal.(*sync.Mutex).Unlock()
-				}
-			}()
-		}
+	if dest.DstNode.DisableConcurrency() {
+		dcNodeId := dest.DstNode.GetId()
+		actual, _ := ec.Graph.ConcurrencyLocks.LoadOrStore(dcNodeId, &sync.Mutex{})
+		mu := actual.(*sync.Mutex)
+		mu.Lock()
+		ec.PendingConcurrencyLocks.Store(dcNodeId, mu)
+		// Fallback: release if ExecuteImpl returns without calling Execute
+		// (end of chain, error, or panic).
+		defer func() {
+			if lockVal, loaded := ec.PendingConcurrencyLocks.LoadAndDelete(dcNodeId); loaded {
+				lockVal.(*sync.Mutex).Unlock()
+			}
+		}()
 	}
 
 	err = dest.DstNode.ExecuteImpl(ec, dest.Port, err)

core/graph.go

@@ -82,7 +82,7 @@ func (ag *ActionGraph) GetEntry() (NodeEntryInterface, error) {
 
 func NewActionGraph() ActionGraph {
 	return ActionGraph{
-		Nodes:         make(map[string]NodeBaseInterface),
+		Nodes:            make(map[string]NodeBaseInterface),
 		ConcurrencyLocks: &sync.Map{},
 	}
 }
@@ -250,10 +250,10 @@ func NewExecutionState(
 	ctx, cancel := context.WithCancel(ctx)
 
 	return &ExecutionState{
-		Graph:                graph,
-		Hierarchy:            make([]NodeBaseInterface, 0),
-		ContextStackLock:     &sync.RWMutex{},
-		OutputCacheLock:      &sync.RWMutex{},
+		Graph:                   graph,
+		Hierarchy:               make([]NodeBaseInterface, 0),
+		ContextStackLock:        &sync.RWMutex{},
+		OutputCacheLock:         &sync.RWMutex{},
 		PendingConcurrencyLocks: &sync.Map{},
 
 		IsDebugSession: debugCb != nil,
@@ -821,6 +821,15 @@ func LoadInputValues(node NodeBaseInterface, nodeI map[string]any, validate bool
 
 	subInputs := map[string][]subInput{}
 
+	// _disable_concurrency is not a regular input, its stored directly on
+	// the node instance so we pull it out before processing the rest.
+	if v, ok := inputValues["_disable_concurrency"]; ok {
+		if b, ok := v.(bool); ok && b {
+			node.SetDisableConcurrency(true)
+		}
+		delete(inputValues, "_disable_concurrency")
+	}
+
 	for portId, inputValue := range inputValues {
 		groupInputId, portIndex, isIndexPort := IsValidIndexPortId(portId)
 		if isIndexPort {

tests_unit/disable_concurrency_input_test.go

@@ -6,47 +6,78 @@ import (
 	"testing"
 
 	"github.com/actionforge/actrun-cli/core"
+	"go.yaml.in/yaml/v4"
 
 	// initialize all nodes
 	_ "github.com/actionforge/actrun-cli/nodes"
 )
 
-func TestDisableConcurrencyInputInjection(t *testing.T) {
-	registries := core.GetRegistries()
-	if len(registries) == 0 {
-		t.Fatal("no node types registered; node factories not loaded")
+func loadTestGraph(t *testing.T, graphStr string) core.ActionGraph {
+	t.Helper()
+	var graphYaml map[string]any
+	if err := yaml.Unmarshal([]byte(graphStr), &graphYaml); err != nil {
+		t.Fatalf("unmarshal YAML: %v", err)
 	}
+	ag, errs := core.LoadGraph(graphYaml, nil, "", false, core.RunOpts{})
+	if len(errs) > 0 {
+		t.Fatalf("LoadGraph: %v", errs[0])
+	}
+	return ag
+}
+
+func TestDisableConcurrencyNotSetByDefault(t *testing.T) {
+	ag := loadTestGraph(t, `
+entry: start
+nodes:
+  - id: start
+    type: core/start@v1
+    position: {x: 0, y: 0}
+  - id: run1
+    type: core/run@v1
+    position: {x: 100, y: 0}
+    inputs:
+      shell: bash
+      script: echo hello
+connections: []
+executions:
+  - src: {node: start, port: exec}
+    dst: {node: run1, port: exec}
+`)
 
-	for id, nodeDef := range registries {
-		// Count execution inputs to determine if this is an execution node.
-		hasExecInput := false
-		for _, inputDef := range nodeDef.Inputs {
-			if inputDef.Exec {
-				hasExecInput = true
-				break
-			}
-		}
-
-		dcDef, hasDC := nodeDef.Inputs[core.InputId("_disable_concurrency")]
-
-		if hasExecInput {
-			if !hasDC {
-				t.Errorf("node %s has execution inputs but is missing _disable_concurrency input", id)
-				continue
-			}
-			if dcDef.Type != "bool" {
-				t.Errorf("node %s: _disable_concurrency type = %q, want \"bool\"", id, dcDef.Type)
-			}
-			if dcDef.Default != false {
-				t.Errorf("node %s: _disable_concurrency default = %v, want false", id, dcDef.Default)
-			}
-			if !dcDef.HideSocket {
-				t.Errorf("node %s: _disable_concurrency HideSocket = false, want true", id)
-			}
-		} else {
-			if hasDC {
-				t.Errorf("node %s has no execution inputs but has _disable_concurrency input", id)
-			}
-		}
+	node := ag.Nodes["run1"]
+	if node == nil {
+		t.Fatal("node run1 not found")
+	}
+	if node.DisableConcurrency() {
+		t.Error("expected DisableConcurrency to be false by default")
+	}
+}
+
+func TestDisableConcurrencySetFromYaml(t *testing.T) {
+	ag := loadTestGraph(t, `
+entry: start
+nodes:
+  - id: start
+    type: core/start@v1
+    position: {x: 0, y: 0}
+  - id: run1
+    type: core/run@v1
+    position: {x: 100, y: 0}
+    inputs:
+      _disable_concurrency: true
+      shell: bash
+      script: echo hello
+connections: []
+executions:
+  - src: {node: start, port: exec}
+    dst: {node: run1, port: exec}
+`)
+
+	node := ag.Nodes["run1"]
+	if node == nil {
+		t.Fatal("node run1 not found")
+	}
+	if !node.DisableConcurrency() {
+		t.Error("expected DisableConcurrency to be true when set in YAML")
 	}
 }