fix: fix thread safety issue in engine invoke (#272)

Author: PeterSchafer

pkg/workflow/engine_test.go

@@ -205,8 +205,8 @@ func Test_Engine_ClonedNetworkAccess(t *testing.T) {
 }
 
 func Test_EngineInvocationConcurrent(t *testing.T) {
-	configuration := configuration.NewInMemory()
-	engine := NewWorkFlowEngine(configuration)
+	config := configuration.NewInMemory()
+	engine := NewWorkFlowEngine(config)
 
 	flagset := pflag.NewFlagSet("1", pflag.ExitOnError)
 	callback := func(invocation InvocationContext, input []Data) ([]Data, error) {
@@ -224,6 +224,9 @@ func Test_EngineInvocationConcurrent(t *testing.T) {
 	stop := make(chan struct{}, N)
 	for range N {
 		go func() {
+			logger := zerolog.Nop()
+			engine.SetLogger(&logger)
+			engine.SetConfiguration(configuration.NewWithOpts())
 			_, invokeErr := engine.Invoke(workflowId)
 			assert.NoError(t, invokeErr)
 			stop <- struct{}{}

pkg/workflow/engineimpl.go

@@ -41,6 +41,8 @@ func (e *EngineImpl) GetLogger() *zerolog.Logger {
 }
 
 func (e *EngineImpl) SetLogger(logger *zerolog.Logger) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.logger = logger
 
 	if e.networkAccess != nil {
@@ -49,6 +51,8 @@ func (e *EngineImpl) SetLogger(logger *zerolog.Logger) {
 }
 
 func (e *EngineImpl) SetConfiguration(config configuration.Configuration) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	e.config = config
 
 	if e.networkAccess != nil {
@@ -247,10 +251,11 @@ func (e *EngineImpl) InvokeWithInputAndConfig(
 
 			// prepare logger
 			prefix := fmt.Sprintf("%s:%d", id.Host, e.invocationCounter)
-			e.mu.Unlock()
-
 			zlogger := e.logger.With().Str("ext", prefix).Logger()
 
+			localUi := e.ui
+			localAnalytics := e.analytics
+
 			// prepare configuration
 			if config == nil {
 				config = e.config.Clone()
@@ -259,9 +264,10 @@ func (e *EngineImpl) InvokeWithInputAndConfig(
 			// prepare networkAccess
 			networkAccess := e.networkAccess.Clone()
 			networkAccess.SetConfiguration(config)
+			e.mu.Unlock()
 
 			// create a context object for the invocation
-			context := NewInvocationContext(id, config, e, networkAccess, zlogger, e.analytics, e.ui)
+			context := NewInvocationContext(id, config, e, networkAccess, zlogger, localAnalytics, localUi)
 
 			// invoke workflow through its callback
 			zlogger.Printf("Workflow Start")