Error handling in chains

component/component.go

@@ -42,6 +42,25 @@ func (c *Component) WithDescription(description string) *Component {
 	return c
 }
 
+// withInputPorts sets input ports collection
+func (c *Component) withInputPorts(collection *port.Collection) *Component {
+	if collection.HasChainError() {
+		return c.WithChainError(collection.ChainError())
+	}
+	c.inputs = collection
+	return c
+}
+
+// withOutputPorts sets input ports collection
+func (c *Component) withOutputPorts(collection *port.Collection) *Component {
+	if collection.HasChainError() {
+		return c.WithChainError(collection.ChainError())
+	}
+
+	c.outputs = collection
+	return c
+}
+
 // WithInputs ads input ports
 func (c *Component) WithInputs(portNames ...string) *Component {
 	if c.HasChainError() {
@@ -52,21 +71,20 @@ func (c *Component) WithInputs(portNames ...string) *Component {
 	if err != nil {
 		return c.WithChainError(err)
 	}
-	c.inputs = c.Inputs().With(ports...)
-	return c
+	return c.withInputPorts(c.Inputs().With(ports...))
 }
 
 // WithOutputs adds output ports
 func (c *Component) WithOutputs(portNames ...string) *Component {
 	if c.HasChainError() {
 		return c
 	}
+
 	ports, err := port.NewGroup(portNames...).Ports()
 	if err != nil {
 		return c.WithChainError(err)
 	}
-	c.outputs = c.Outputs().With(ports...)
-	return c
+	return c.withOutputPorts(c.Outputs().With(ports...))
 }
 
 // WithInputsIndexed creates multiple prefixed ports
@@ -75,8 +93,7 @@ func (c *Component) WithInputsIndexed(prefix string, startIndex int, endIndex in
 		return c
 	}
 
-	c.inputs = c.Inputs().WithIndexed(prefix, startIndex, endIndex)
-	return c
+	return c.withInputPorts(c.Inputs().WithIndexed(prefix, startIndex, endIndex))
 }
 
 // WithOutputsIndexed creates multiple prefixed ports
@@ -85,8 +102,7 @@ func (c *Component) WithOutputsIndexed(prefix string, startIndex int, endIndex i
 		return c
 	}
 
-	c.outputs = c.Outputs().WithIndexed(prefix, startIndex, endIndex)
-	return c
+	return c.withOutputPorts(c.Outputs().WithIndexed(prefix, startIndex, endIndex))
 }
 
 // WithActivationFunc sets activation function
@@ -130,8 +146,7 @@ func (c *Component) Outputs() *port.Collection {
 func (c *Component) OutputByName(name string) *port.Port {
 	outputPort := c.Outputs().ByName(name)
 	if outputPort.HasChainError() {
-		c.SetChainError(outputPort.ChainError())
-		return nil
+		return port.New("").WithChainError(outputPort.ChainError())
 	}
 	return outputPort
 }
@@ -143,7 +158,7 @@ func (c *Component) InputByName(name string) *port.Port {
 	}
 	inputPort := c.Inputs().ByName(name)
 	if inputPort.HasChainError() {
-		c.SetChainError(inputPort.ChainError())
+		return port.New("").WithChainError(inputPort.ChainError())
 	}
 	return inputPort
 }
@@ -157,25 +172,38 @@ func (c *Component) hasActivationFunction() bool {
 	return c.f != nil
 }
 
-// MaybeActivate tries to run the activation function if all required conditions are met
-// @TODO: hide this method from user
-// @TODO: can we remove named return ?
-func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
-	//Bubble up chain errors from ports
+// propagateChainErrors propagates up all chain errors that might have not been propagated yet
+func (c *Component) propagateChainErrors() {
+	if c.Inputs().HasChainError() {
+		c.SetChainError(c.Inputs().ChainError())
+		return
+	}
+
+	if c.Outputs().HasChainError() {
+		c.SetChainError(c.Outputs().ChainError())
+		return
+	}
+
 	for _, p := range c.Inputs().PortsOrNil() {
 		if p.HasChainError() {
-			c.Inputs().SetChainError(p.ChainError())
-			c.SetChainError(c.Inputs().ChainError())
-			break
+			c.SetChainError(p.ChainError())
+			return
 		}
 	}
+
 	for _, p := range c.Outputs().PortsOrNil() {
 		if p.HasChainError() {
-			c.Outputs().SetChainError(p.ChainError())
-			c.SetChainError(c.Outputs().ChainError())
-			break
+			c.SetChainError(p.ChainError())
+			return
 		}
 	}
+}
+
+// MaybeActivate tries to run the activation function if all required conditions are met
+// @TODO: hide this method from user
+// @TODO: can we remove named return ?
+func (c *Component) MaybeActivate() (activationResult *ActivationResult) {
+	c.propagateChainErrors()
 
 	if c.HasChainError() {
 		activationResult = NewActivationResult(c.Name()).WithChainError(c.ChainError())

component/component_test.go

@@ -41,44 +41,42 @@ func TestNewComponent(t *testing.T) {
 }
 
 func TestComponent_FlushOutputs(t *testing.T) {
-	sink := port.New("sink")
-
-	componentWithNoOutputs := New("c1")
-	componentWithCleanOutputs := New("c1").WithOutputs("o1", "o2")
-
-	componentWithAllOutputsSet := New("c1").WithOutputs("o1", "o2")
-	componentWithAllOutputsSet.Outputs().ByNames("o1").PutSignals(signal.New(777))
-	componentWithAllOutputsSet.Outputs().ByNames("o2").PutSignals(signal.New(888))
-	componentWithAllOutputsSet.Outputs().ByNames("o1", "o2").PipeTo(sink)
-
 	tests := []struct {
-		name       string
-		component  *Component
-		destPort   *port.Port //Where the component flushes ALL it's inputs
-		assertions func(t *testing.T, componentAfterFlush *Component, destPort *port.Port)
+		name         string
+		getComponent func() *Component
+		assertions   func(t *testing.T, componentAfterFlush *Component)
 	}{
 		{
-			name:      "no outputs",
-			component: componentWithNoOutputs,
-			destPort:  nil,
-			assertions: func(t *testing.T, componentAfterFlush *Component, destPort *port.Port) {
+			name: "no outputs",
+			getComponent: func() *Component {
+				return New("c1")
+			},
+			assertions: func(t *testing.T, componentAfterFlush *Component) {
 				assert.NotNil(t, componentAfterFlush.Outputs())
 				assert.Zero(t, componentAfterFlush.Outputs().Len())
 			},
 		},
 		{
-			name:      "output has no signal set",
-			component: componentWithCleanOutputs,
-			destPort:  nil,
-			assertions: func(t *testing.T, componentAfterFlush *Component, destPort *port.Port) {
+			name: "output has no signal set",
+			getComponent: func() *Component {
+				return New("c1").WithOutputs("o1", "o2")
+			},
+			assertions: func(t *testing.T, componentAfterFlush *Component) {
 				assert.False(t, componentAfterFlush.Outputs().AnyHasSignals())
 			},
 		},
 		{
-			name:      "happy path",
-			component: componentWithAllOutputsSet,
-			destPort:  sink,
-			assertions: func(t *testing.T, componentAfterFlush *Component, destPort *port.Port) {
+			name: "happy path",
+			getComponent: func() *Component {
+				sink := port.New("sink")
+				c := New("c1").WithOutputs("o1", "o2")
+				c.Outputs().ByNames("o1").PutSignals(signal.New(777))
+				c.Outputs().ByNames("o2").PutSignals(signal.New(888))
+				c.Outputs().ByNames("o1", "o2").PipeTo(sink)
+				return c
+			},
+			assertions: func(t *testing.T, componentAfterFlush *Component) {
+				destPort := componentAfterFlush.OutputByName("o1").Pipes().PortsOrNil()[0]
 				allPayloads, err := destPort.AllSignalsPayloads()
 				assert.NoError(t, err)
 				assert.Contains(t, allPayloads, 777)
@@ -88,11 +86,25 @@ func TestComponent_FlushOutputs(t *testing.T) {
 				assert.False(t, componentAfterFlush.Outputs().AnyHasSignals())
 			},
 		},
+		{
+			name: "with chain error",
+			getComponent: func() *Component {
+				sink := port.New("sink")
+				c := New("c").WithOutputs("o1").WithChainError(errors.New("some error"))
+				//Lines below are ignored as error immediately propagates up to component level
+				c.Outputs().ByName("o1").PipeTo(sink)
+				c.Outputs().ByName("o1").PutSignals(signal.New("signal from component with chain error"))
+				return c
+			},
+			assertions: func(t *testing.T, componentAfterFlush *Component) {
+				assert.False(t, componentAfterFlush.OutputByName("o1").HasPipes())
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tt.component.FlushOutputs()
-			tt.assertions(t, tt.component, tt.destPort)
+			componentAfter := tt.getComponent().FlushOutputs()
+			tt.assertions(t, componentAfter)
 		})
 	}
 }
@@ -481,6 +493,28 @@ func TestComponent_MaybeActivate(t *testing.T) {
 				err:           NewErrWaitForInputs(true),
 			},
 		},
+		{
+			name: "with chain error from input port",
+			getComponent: func() *Component {
+				c := New("c").WithInputs("i1").WithOutputs("o1")
+				c.Inputs().With(port.New("p").WithChainError(errors.New("some error")))
+				return c
+			},
+			wantActivationResult: NewActivationResult("c").
+				WithActivationCode(ActivationCodeUndefined).
+				WithChainError(errors.New("some error")),
+		},
+		{
+			name: "with chain error from output port",
+			getComponent: func() *Component {
+				c := New("c").WithInputs("i1").WithOutputs("o1")
+				c.Outputs().With(port.New("p").WithChainError(errors.New("some error")))
+				return c
+			},
+			wantActivationResult: NewActivationResult("c").
+				WithActivationCode(ActivationCodeUndefined).
+				WithChainError(errors.New("some error")),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -630,3 +664,59 @@ func TestComponent_WithLabels(t *testing.T) {
 		})
 	}
 }
+
+func TestComponent_ShortcutMethods(t *testing.T) {
+	t.Run("InputByName", func(t *testing.T) {
+		c := New("c").WithInputs("a", "b", "c")
+		assert.Equal(t, port.New("b"), c.InputByName("b"))
+	})
+
+	t.Run("OutputByName", func(t *testing.T) {
+		c := New("c").WithOutputs("a", "b", "c")
+		assert.Equal(t, port.New("b"), c.OutputByName("b"))
+	})
+}
+
+func TestComponent_ClearInputs(t *testing.T) {
+	tests := []struct {
+		name         string
+		getComponent func() *Component
+		assertions   func(t *testing.T, componentAfter *Component)
+	}{
+		{
+			name: "no side effects",
+			getComponent: func() *Component {
+				return New("c").WithInputs("i1").WithOutputs("o1")
+			},
+			assertions: func(t *testing.T, componentAfter *Component) {
+				assert.Equal(t, 1, componentAfter.Inputs().Len())
+				assert.Equal(t, 1, componentAfter.Outputs().Len())
+				assert.False(t, componentAfter.Inputs().AnyHasSignals())
+				assert.False(t, componentAfter.Outputs().AnyHasSignals())
+			},
+		},
+		{
+			name: "only inputs are cleared",
+			getComponent: func() *Component {
+				c := New("c").WithInputs("i1").WithOutputs("o1")
+				c.Inputs().ByName("i1").PutSignals(signal.New(10))
+				c.Outputs().ByName("o1").PutSignals(signal.New(20))
+				return c
+			},
+			assertions: func(t *testing.T, componentAfter *Component) {
+				assert.Equal(t, 1, componentAfter.Inputs().Len())
+				assert.Equal(t, 1, componentAfter.Outputs().Len())
+				assert.False(t, componentAfter.Inputs().AnyHasSignals())
+				assert.True(t, componentAfter.Outputs().ByName("o1").HasSignals())
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			componentAfter := tt.getComponent().ClearInputs()
+			if tt.assertions != nil {
+				tt.assertions(t, componentAfter)
+			}
+		})
+	}
+}

fmesh.go

@@ -87,22 +87,22 @@ func (fm *FMesh) WithConfig(config Config) *FMesh {
 }
 
 // runCycle runs one activation cycle (tries to activate ready components)
-func (fm *FMesh) runCycle() (*cycle.Cycle, error) {
+func (fm *FMesh) runCycle() *cycle.Cycle {
+	newCycle := cycle.New()
+
 	if fm.HasChainError() {
-		return nil, fm.ChainError()
+		return newCycle.WithChainError(fm.ChainError())
 	}
 
 	if fm.Components().Len() == 0 {
-		return nil, errors.New("failed to run cycle: no components found")
+		return newCycle.WithChainError(errors.New("failed to run cycle: no components found"))
 	}
 
-	newCycle := cycle.New()
-
 	var wg sync.WaitGroup
 
 	components, err := fm.Components().Components()
 	if err != nil {
-		return nil, fmt.Errorf("failed to run cycle: %w", err)
+		return newCycle.WithChainError(fmt.Errorf("failed to run cycle: %w", err))
 	}
 
 	for _, c := range components {
@@ -130,7 +130,7 @@ func (fm *FMesh) runCycle() (*cycle.Cycle, error) {
 		}
 	}
 
-	return newCycle, nil
+	return newCycle
 }
 
 // DrainComponents drains the data from activated components
@@ -188,13 +188,7 @@ func (fm *FMesh) Run() (cycle.Cycles, error) {
 	allCycles := cycle.NewGroup()
 	cycleNumber := 0
 	for {
-		cycleResult, err := fm.runCycle()
-
-		if err != nil {
-			return nil, err
-		}
-
-		cycleResult.WithNumber(cycleNumber)
+		cycleResult := fm.runCycle().WithNumber(cycleNumber)
 
 		if cycleResult.HasChainError() {
 			fm.SetChainError(cycleResult.ChainError())
@@ -216,7 +210,7 @@ func (fm *FMesh) Run() (cycle.Cycles, error) {
 			return cycles, stopError
 		}
 
-		err = fm.drainComponents(cycleResult)
+		err := fm.drainComponents(cycleResult)
 		if err != nil {
 			return nil, err
 		}