IWF-132: Use firstRunId instead workflowId when naming WaitForStateExecutionCompletion workflow #447

lwolczynski · 2024-10-11T23:21:56Z

Description

Checklist

Code compiles correctly
Tests for the changes have been added
All tests passing
This PR change is backwards-compatible
This PR CONTAINS a (planned) breaking change (it is not backwards compatible)

Related Issue

Closes #404

…ecutionCompletion workflow

longquanzheng · 2024-10-11T23:36:40Z

service/api/service.go

+	}
+
+	if response.FirstRunId == "" {
+		parentWfId = req.WorkflowId // Cadence


If we know cadence doesn't have this, we can avoid calling the DescribeWorkflowExecution API for cadence.

Although this API is really fast (it's backed by caching on cadence/temporal) but still nice to do so if this is not too much. (we do have cadence users in OSS, some ppl just perfer using cadence)

Added GetBackendType method to UnifiedClient

longquanzheng · 2024-10-11T23:42:54Z

service/api/service.go

 	}

 	options := uclient.StartWorkflowOptions{
-		ID:        workflowId,
+		// TODO: Switch currentWorkflowId to _ (after renaming) after a short amount of time to ensure backward compatibility


Ohh interesting! I was thinking of making two deployment -- first deploy the workflowImpl code (which signalWIthStart both), after some time, deploy the api service.
But what you put here made me feel like it's nice to use a config?
Potentially using the Api ApiConfig yaml:"api"`` in https://github.com/indeedeng/iwf/blob/main/config/config.go

waitForStateCompletionMigration: SignalWithStartOn: both/new/old WaitForOn:old/new

In workflowImpl, we have access to ApiConfig via env like this: https://github.com/indeedeng/iwf/blob/main/service/interpreter/continueAsNewer.go#L57

That's a good idea. I'll implement it

longquanzheng · 2024-10-11T23:43:40Z

service/api/service.go

+	if response.FirstRunId == "" {
+		parentWfId = req.WorkflowId // Cadence
+	} else {
+		parentWfId = response.FirstRunId // Temporal


very nit: parentId? it's not workflowId Lol.

service/interpreter/workflowImpl.go

longquanzheng · 2024-10-11T23:46:24Z

service/interpreter/workflowImpl.go

+
+		if response.FirstRunId != "" { // Temporal
+			// Start WaitForStateCompletionWorkflow with a new name to ensure smooth transition
+			err = unifiedClient.SignalWithStartWaitForStateCompletionWorkflow(


Nit: we can refactor the code to a common method

longquanzheng · 2024-10-14T20:11:39Z

service/interpreter/workflowImpl.go

+	if state.WaitForKey != nil {
+		return service.IwfSystemConstPrefix + parentId + "_" + state.StateId + "_" + *state.WaitForKey
+	} else {
+		return service.IwfSystemConstPrefix + parentId + "_" + *executionContext.StateExecutionId
+	}


Nit: it would be nice to move this to a common place to share with api service: GetWorkflowIdForWaitForStateExecution(parentId, stateExeId, waitForKey, stateId)

longquanzheng · 2024-10-14T20:18:34Z

service/interpreter/workflowImpl.go

+		// panic will let the workflow task will retry until the signal is sent
+		panic(fmt.Errorf("failed to signal on completion %w", err))
+	}
+	return err


I think this func doesn't need to return anything(error).

It's a bit tricky here that we need to expand on later about "non determinism and how workflow task is executed in Cadence/Temporal".

For here, we intentionally swallow the error of "alreadyStarted", and panic on any other errors. By panicing, it will just retry automatically.

We don't want to proceed to other states for any other errors. The proceed to the state is the feature for workflow state failure.

Yes, I think this fails one of integration tests and I'm looking into it atm

Fixed the issue

service/api/service.go

lwolczynski · 2024-10-14T21:18:12Z

service/interpreter/workflowImpl.go

+		if provider.GetBackendType() == service.BackendTypeCadence ||
+			(provider.GetBackendType() == service.BackendTypeTemporal && (signalWithStartOn == "old" || signalWithStartOn == "both")) {
+			workflowId := utils.GetWorkflowIdForWaitForStateExecution(provider.GetWorkflowInfo(ctx).FirstRunID, *executionContext.StateExecutionId, *state.WaitForKey, state.StateId)
+
+			err = signalWithStart(unifiedClient, workflowId)
+			if err != nil && !unifiedClient.IsWorkflowAlreadyStartedError(err) {
+				// WorkflowAlreadyStartedError is returned when the started workflow is closed and the signal is not sent
+				// panic will let the workflow task will retry until the signal is sent
+				panic(fmt.Errorf("failed to signal on completion %w", err))
+			}
 		}

-		err := unifiedClient.SignalWithStartWaitForStateCompletionWorkflow(
-			context.Background(),
-			uclient.StartWorkflowOptions{
-				ID:                       workflowId,
-				TaskQueue:                env.GetTaskQueue(),
-				WorkflowExecutionTimeout: 60 * time.Second, // timeout doesn't matter here as it will complete immediate with the signal
-			},
-			iwfidl.StateCompletionOutput{})
-		if err != nil && !unifiedClient.IsWorkflowAlreadyStartedError(err) {
-			// WorkflowAlreadyStartedError is returned when the started workflow is closed and the signal is not sent
-			// panic will let the workflow task will retry until the signal is sent
-			panic(fmt.Errorf("failed to signal on completion %w", err))
+		// signalWithStart with new workflowId (containing firstRunId)
+		if provider.GetBackendType() == service.BackendTypeTemporal && (signalWithStartOn == "both" || signalWithStartOn == "new") {
+			workflowId := utils.GetWorkflowIdForWaitForStateExecution(provider.GetWorkflowInfo(ctx).FirstRunID, *executionContext.StateExecutionId, *state.WaitForKey, state.StateId)
+
+			// Start WaitForStateCompletionWorkflow with a new name to ensure smooth transition
+			err = signalWithStart(unifiedClient, workflowId)
+			if err != nil && !unifiedClient.IsWorkflowAlreadyStartedError(err) {
+				// WorkflowAlreadyStartedError is returned when the started workflow is closed and the signal is not sent
+				// panic will let the workflow task will retry until the signal is sent
+				panic(fmt.Errorf("failed to signal on completion %w", err))
+			}
 		}
 	}


Refactored this from two nested if statements into two separate ifs with more complicated expressions

longquanzheng · 2024-10-14T21:19:12Z

service/api/service.go

+			parentId = req.WorkflowId
+		}
+
+		workflowId = utils.GetWorkflowIdForWaitForStateExecution(parentId, *req.StateExecutionId, *req.WaitForKey, *req.StateId)


*req.WaitForKey will assume that field is always present (otherwise it will be a panic -- nil pointer error). I think only StateExecutionId is required. So we probably need to pass pointer into the method and let the method do a nil check

…tionMigration

longquanzheng · 2024-10-14T22:49:35Z

service/interpreter/workflowImpl.go

+			if err != nil && !unifiedClient.IsWorkflowAlreadyStartedError(err) {
+				// WorkflowAlreadyStartedError is returned when the started workflow is closed and the signal is not sent
+				// panic will let the workflow task will retry until the signal is sent
+				panic(fmt.Errorf("failed to signal on completion %w", err))


Nit: it will look slightly better to move this error handling into the signalWithStart method

longquanzheng · 2024-10-14T23:05:43Z

config/config.go

+func GetSignalWithStartOnWithDefault(config Config) string {
+	if config.Api.WaitForStateCompletionMigration.SignalWithStartOn != "" {
+		return config.Api.WaitForStateCompletionMigration.SignalWithStartOn
+	}
+	return "old"
+}
+
+func GetWaitForOnWithDefault(config Config) string {


Nit: I actually should have done it this using struct member method instead:

func (c Config) GetSignalWithStartOnWithDefault() string{ // the same code here }

And to use it:

config.GetSignalWithStartOnWithDefault()

This is very common pattern in Golang, I don't know why I didn't do that way 😂

longquanzheng · 2024-10-15T17:57:38Z

config/config.go

@@ -31,6 +31,15 @@ type (
 		// omitRpcInputOutputInHistory is the flag to omit rpc input/output in history
 		// the input/output is only for debugging purpose but could be too expensive to store
 		OmitRpcInputOutputInHistory *bool `yaml:"omitRpcInputOutputInHistory"`
+		// WaitForStateCompletionMigration is used to control naming of the continuedAsNew workflows


typo: not continuedAsNew

IWF-132: Use firstRunId instead workflowId when naming WaitForStateEx…

8bf1dc3

…ecutionCompletion workflow

lwolczynski requested a review from longquanzheng October 11, 2024 23:22

longquanzheng reviewed Oct 11, 2024

View reviewed changes

service/interpreter/workflowImpl.go Outdated Show resolved Hide resolved

longquanzheng reviewed Oct 11, 2024

View reviewed changes

lwolczynski added 3 commits October 14, 2024 09:42

IWF-132: Rename variable

03660b4

IWF-132: Add FirstRunID to WorkflowInfo

828af2b

IWF-132: Add GetBackendType method to UnifiedClient

dbcf2a4

lwolczynski force-pushed the jira/lwolczynski/IRLWEB-132 branch from f7df0d5 to dbcf2a4 Compare October 14, 2024 17:18

IWF-132: Add WaitForStateCompletionMigration config

6fd76dd

lwolczynski requested a review from longquanzheng October 14, 2024 19:28

longquanzheng reviewed Oct 14, 2024

View reviewed changes

service/api/service.go Outdated Show resolved Hide resolved

lwolczynski added 2 commits October 14, 2024 15:57

IWF-132: Refactor

20bf566

IWF-132: Refactor

3afb3e5

lwolczynski force-pushed the jira/lwolczynski/IRLWEB-132 branch from 6cc92ab to 3afb3e5 Compare October 14, 2024 21:16

lwolczynski commented Oct 14, 2024

View reviewed changes

lwolczynski requested a review from longquanzheng October 14, 2024 21:18

longquanzheng reviewed Oct 14, 2024

View reviewed changes

lwolczynski added 3 commits October 14, 2024 16:37

IWF-132: Fix GetWorkflowIdForWaitForStateExecution

684a6a0

IWF-132: Add helpers that return default value for WaitForStateComple…

a7ce6eb

…tionMigration

IWF-132: Fix

8d401e5

longquanzheng reviewed Oct 14, 2024

View reviewed changes

longquanzheng approved these changes Oct 14, 2024

View reviewed changes

longquanzheng reviewed Oct 14, 2024

View reviewed changes

lwolczynski added 2 commits October 15, 2024 11:32

IWF-132: Refactor

aaac33f

IWF-132: Refactor

15def77

lwolczynski force-pushed the jira/lwolczynski/IRLWEB-132 branch from a3f4e52 to 15def77 Compare October 15, 2024 17:13

longquanzheng reviewed Oct 15, 2024

View reviewed changes

IWF-132: Fix typo

be429c5

lwolczynski merged commit 2181748 into main Oct 15, 2024
5 checks passed

lwolczynski deleted the jira/lwolczynski/IRLWEB-132 branch October 18, 2024 20:03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

IWF-132: Use firstRunId instead workflowId when naming WaitForStateExecutionCompletion workflow #447

IWF-132: Use firstRunId instead workflowId when naming WaitForStateExecutionCompletion workflow #447

lwolczynski commented Oct 11, 2024 •

edited

Loading

longquanzheng Oct 11, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 11, 2024

longquanzheng Oct 11, 2024

lwolczynski Oct 14, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 11, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 11, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 14, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 14, 2024 •

edited

Loading

lwolczynski Oct 14, 2024

lwolczynski Oct 14, 2024

lwolczynski Oct 14, 2024

longquanzheng Oct 14, 2024 •

edited

Loading

lwolczynski Oct 14, 2024

longquanzheng Oct 14, 2024

lwolczynski Oct 15, 2024

longquanzheng Oct 14, 2024

lwolczynski Oct 15, 2024

longquanzheng Oct 15, 2024

IWF-132: Use firstRunId instead workflowId when naming WaitForStateExecutionCompletion workflow #447

IWF-132: Use firstRunId instead workflowId when naming WaitForStateExecutionCompletion workflow #447

Conversation

lwolczynski commented Oct 11, 2024 • edited Loading

Description

Checklist

Related Issue

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

longquanzheng Oct 14, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

longquanzheng Oct 14, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lwolczynski commented Oct 11, 2024 •

edited

Loading

longquanzheng Oct 14, 2024 •

edited

Loading

longquanzheng Oct 14, 2024 •

edited

Loading