Repository: honeydipper/honeydipper
Pull request #512 (2023-02-12 18:48:38)

fix(core): resume_token should be unique among cluster members
#### Description
Prepend the daemonID in front of the resume_token makes it globally unique. This
avoids token collision, and ensures the right session gets resumed, when called.

#### This PR fixes the following issues
N/A



Patch:
diff --git a/internal/service/engine.go b/internal/service/engine.go
index ecea41b..94a7406 100644
--- a/internal/service/engine.go
+++ b/internal/service/engine.go
@@ -48,6 +48,11 @@ func (h *WorkflowHelper) GetConfig() *config.Config {
 	return h.engine.config
 }
 
+// GetDaemonID method gets an identifier to the current daemon.
+func (h *WorkflowHelper) GetDaemonID() string {
+	return h.engine.daemonID
+}
+
 // StartEngine Starts the engine service.
 func StartEngine(cfg *config.Config) {
 	engine = NewService(cfg, "engine")
diff --git a/internal/service/service.go b/internal/service/service.go
index ecb69bb..fe39abb 100644
--- a/internal/service/service.go
+++ b/internal/service/service.go
@@ -75,6 +75,7 @@ type Service struct {
 	ResponseFactory    *api.ResponseFactory
 	healthy            bool
 	drainingGroup      *sync.WaitGroup
+	daemonID           string
 }
 
 var (
@@ -92,6 +93,7 @@ var (
 func NewService(cfg *config.Config, name string) *Service {
 	svc := &Service{
 		name:           name,
+		daemonID:       dipper.GetIP(),
 		config:         cfg,
 		driverRuntimes: map[string]*driver.Runtime{},
 		expects:        map[string][]ExpectHandler{},
@@ -716,7 +718,7 @@ func handleAPI(from *driver.Runtime, m *dipper.Message) {
 
 func handleReload(from *driver.Runtime, m *dipper.Message) {
 	daemonID, ok := m.Labels["daemonID"]
-	if ok && daemonID != dipper.GetIP() {
+	if ok && daemonID != Services[from.Service].daemonID {
 		return
 	}
 
diff --git a/internal/workflow/execute.go b/internal/workflow/execute.go
index e41fee1..982182b 100644
--- a/internal/workflow/execute.go
+++ b/internal/workflow/execute.go
@@ -106,7 +106,7 @@ func (w *Session) executeRound(msg *dipper.Message) {
 		if w.ctx == nil {
 			w.ctx = map[string]interface{}{}
 		}
-		w.ctx["resume_token"] = "/" + w.workflow.Name + "/" + w.ID
+		w.ctx["resume_token"] = w.store.Helper.GetDaemonID() + "/" + w.workflow.Name + "/" + w.ID
 	}
 
 	w.executeIteration(msg)
diff --git a/internal/workflow/mock_workflow/store.go b/internal/workflow/mock_workflow/store.go
index e0aaac1..2d40f43 100644
--- a/internal/workflow/mock_workflow/store.go
+++ b/internal/workflow/mock_workflow/store.go
@@ -56,6 +56,20 @@ func (mr *MockSessionStoreHelperMockRecorder) GetConfig() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetConfig", reflect.TypeOf((*MockSessionStoreHelper)(nil).GetConfig))
 }
 
+// GetDaemonID mocks base method.
+func (m *MockSessionStoreHelper) GetDaemonID() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetDaemonID")
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// GetDaemonID indicates an expected call of GetDaemonID.
+func (mr *MockSessionStoreHelperMockRecorder) GetDaemonID() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDaemonID", reflect.TypeOf((*MockSessionStoreHelper)(nil).GetDaemonID))
+}
+
 // SendMessage mocks base method.
 func (m *MockSessionStoreHelper) SendMessage(msg *dipper.Message) {
 	m.ctrl.T.Helper()
diff --git a/internal/workflow/store.go b/internal/workflow/store.go
index 842a53b..eae31bd 100644
--- a/internal/workflow/store.go
+++ b/internal/workflow/store.go
@@ -20,6 +20,7 @@ import (
 type SessionStoreHelper interface {
 	GetConfig() *config.Config
 	SendMessage(msg *dipper.Message)
+	GetDaemonID() string
 }
 
 // SessionStore stores session in memory and provides helper function for session to perform.


Test patch:
diff --git a/internal/workflow/condition_test.go b/internal/workflow/condition_test.go
index ab65d4b..b07352d 100644
--- a/internal/workflow/condition_test.go
+++ b/internal/workflow/condition_test.go
@@ -29,6 +29,7 @@ func TestConditionElse(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -77,6 +78,7 @@ func TestSwitchDefault(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -127,6 +129,7 @@ func TestSwitch(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -186,6 +189,7 @@ func TestWorkflowLoop(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
diff --git a/internal/workflow/session_context_test.go b/internal/workflow/session_context_test.go
index 53ad705..fc3bde4 100644
--- a/internal/workflow/session_context_test.go
+++ b/internal/workflow/session_context_test.go
@@ -127,6 +127,7 @@ func TestWorkflowWithEventContext(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -195,6 +196,7 @@ func TestWorkflowWithEventSectionInContext(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -268,6 +270,7 @@ func TestWorkflowWithNamedContext(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
diff --git a/internal/workflow/session_faulttolerant_test.go b/internal/workflow/session_faulttolerant_test.go
index dffd032..cf9890b 100644
--- a/internal/workflow/session_faulttolerant_test.go
+++ b/internal/workflow/session_faulttolerant_test.go
@@ -56,6 +56,7 @@ func TestWorkflowErrorWorkflowNotDefined(t *testing.T) {
 		"ctx":      map[string]interface{}{},
 		"steps":    []map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 	}
@@ -92,6 +93,7 @@ func TestWorkflowIterateEmptyAsChild(t *testing.T) {
 		"ctx":   map[string]interface{}{},
 		"steps": []map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 	}
diff --git a/internal/workflow/session_hook_test.go b/internal/workflow/session_hook_test.go
index 78ec228..b49dc0d 100644
--- a/internal/workflow/session_hook_test.go
+++ b/internal/workflow/session_hook_test.go
@@ -34,6 +34,7 @@ func TestOnSessionHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -68,6 +69,9 @@ func TestOnSessionHook(t *testing.T) {
 					},
 				},
 				"ctx": map[string]interface{}{},
+				"asserts": func() {
+					mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
+				},
 			},
 		},
 	}
@@ -85,6 +89,7 @@ func TestOnFirstActionHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -153,6 +158,7 @@ func TestOnExitHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -205,6 +211,7 @@ func TestOnSuccessHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -259,6 +266,7 @@ func TestOnFailureHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -360,6 +368,7 @@ func TestOnErrorHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -458,6 +467,7 @@ func TestFailureInHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -508,6 +518,7 @@ func TestErrorInCompletionHook(t *testing.T) {
 			},
 		},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
diff --git a/internal/workflow/session_parallel_test.go b/internal/workflow/session_parallel_test.go
index ce472d6..e8ba3cb 100644
--- a/internal/workflow/session_parallel_test.go
+++ b/internal/workflow/session_parallel_test.go
@@ -65,6 +65,7 @@ func TestWorkflowIterateParallel(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(DipperMsgEq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -188,6 +189,7 @@ func TestWorkflowThreads(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(DipperMsgEq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
diff --git a/internal/workflow/store_session_test.go b/internal/workflow/store_session_test.go
index f932570..da1b642 100644
--- a/internal/workflow/store_session_test.go
+++ b/internal/workflow/store_session_test.go
@@ -49,6 +49,7 @@ func TestWorkflowNoopAsChild(t *testing.T) {
 		"ctx":   map[string]interface{}{},
 		"steps": []map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 	}
@@ -68,6 +69,7 @@ func TestWorkflowNoop(t *testing.T) {
 		"ctx":   map[string]interface{}{},
 		"steps": []map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 	}
@@ -81,6 +83,7 @@ func TestCallWorkflow(t *testing.T) {
 		"ctx":      map[string]interface{}{},
 		"steps":    []map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 	}
@@ -93,6 +96,7 @@ func TestCallDriver(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -142,6 +146,7 @@ func TestCallFunction(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -202,6 +207,7 @@ func TestWorkflowSteps(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -302,6 +308,7 @@ func TestWorkflowResumeWithTimeout(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -367,6 +374,7 @@ func TestWorkflowResume(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -438,6 +446,7 @@ func TestContinueNonexistSession(t *testing.T) {
 		"msg":      &dipper.Message{},
 		"ctx":      map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Any()).Times(0)
 		},
 		"steps": []map[string]interface{}{
@@ -479,6 +488,7 @@ func TestWorkflowResumeCrash(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",
@@ -558,6 +568,7 @@ func TestWorkflowIterate(t *testing.T) {
 		"msg": &dipper.Message{},
 		"ctx": map[string]interface{}{},
 		"asserts": func() {
+			mockHelper.EXPECT().GetDaemonID().AnyTimes().Return("")
 			mockHelper.EXPECT().SendMessage(gomock.Eq(&dipper.Message{
 				Channel: "eventbus",
 				Subject: "command",