Use MutableSideEffect to access dynamic config from inside workflows #7268

carlydf · 2025-02-06T04:15:45Z

What changed?

Use MutableSideEffect to access dynamic config from inside workflows.

Why?

To prevent non-determinism in the workflow code.
To allow the workflows to use the latest dynamic config values at all times.

How did you test it?

Tests are failing for other reasons, so I'm unable to test until I pull in those changes.

Potential risks

Documentation

Is hotfix candidate?

dnr · 2025-02-06T17:58:15Z

service/worker/workerdeployment/drainage_workflow.go

+	defaultVisibilityGrace   = 3 * time.Minute
+)
+
+func DrainageWorkflowWithDC(dc *dynamicconfig.Collection, ns string) func(ctx workflow.Context, version *deploymentspb.WorkerDeploymentVersion, first bool) error {
 	return func(ctx workflow.Context, version *deploymentspb.WorkerDeploymentVersion, first bool) error {


I would prefer not writing the whole workflow in an anonymous function inside the main one. Just return a closure that calls the workflow function with the config so it's still at the top level, and the thing that binds in the config can be mostly ignored.

dnr · 2025-02-06T17:58:36Z

service/worker/workerdeployment/drainage_workflow.go

+	defaultVisibilityGrace   = 3 * time.Minute
+)
+
+func DrainageWorkflowWithDC(dc *dynamicconfig.Collection, ns string) func(ctx workflow.Context, version *deploymentspb.WorkerDeploymentVersion, first bool) error {


Here and in the other workflows too: Instead of passing in the whole Collection, I think it would be good to follow the usual pattern of making a Config struct that has only the 1-2 needed values as functions, already bound to the Collection. It's easier to test that way. (Though I have mixed feelings on this one, when it gets big enough I think the Config struct pattern gets annoying)

dnr · 2025-02-06T17:59:17Z

service/worker/workerdeployment/drainage_workflow.go

+	defaultVisibilityGrace   = 3 * time.Minute
+)
+
+func DrainageWorkflowWithDC(dc *dynamicconfig.Collection, ns string) func(ctx workflow.Context, version *deploymentspb.WorkerDeploymentVersion, first bool) error {


Here and in the other workflows too: The config or dc or whatever parameter/field it is should be named something like unsafeConfig or nondeterministicConfig or something better, and with a big comment, to make it clear that you can't just use it from the workflow, it has to be done in a [mutable]sideeffect. I feel like it's inviting bugs otherwise

dnr · 2025-02-06T18:00:13Z

service/worker/workerdeployment/drainage_workflow.go

 		}
 	}
 }
+
+func getRefreshInterval(ctx workflow.Context, dc *dynamicconfig.Collection, ns string) (time.Duration, error) {


Can we consolidate all these so there's one mutablesideeffect that returns a struct of all the needed values?

dnr · 2025-02-06T18:01:11Z

service/worker/workerdeployment/drainage_workflow.go

+	aDur, ok := a.(time.Duration)
+	if !ok {
+		return false
+	}
+	bDur, ok := b.(time.Duration)
+	if !ok {
+		return false
+	}
+	return aDur == bDur


does this not work?

Suggested change

aDur, ok := a.(time.Duration)

if !ok {

return false

}

bDur, ok := b.(time.Duration)

if !ok {

return false

}

return aDur == bDur

return a == b

dnr · 2025-02-06T18:01:39Z

service/worker/workerdeployment/drainage_workflow.go

 		}
 	}
 }
+
+func getRefreshInterval(ctx workflow.Context, dc *dynamicconfig.Collection, ns string) (time.Duration, error) {
+	get := func(ctx workflow.Context) interface{} {


use any everywhere, it's more readable

dnr · 2025-02-06T18:04:07Z

service/worker/workerdeployment/workflow.go

@@ -154,7 +161,7 @@ func (d *WorkflowRunner) run(ctx workflow.Context) error {
 	// Note, if update requests come in faster than they
 	// are handled, there will not be a moment where the workflow has
 	// nothing pending which means this will run forever.
-	return workflow.NewContinueAsNewError(ctx, Workflow, d.WorkerDeploymentWorkflowArgs)
+	return workflow.NewContinueAsNewError(ctx, WorkflowWithDC(d.dc), d.WorkerDeploymentWorkflowArgs)


Use the name here, I'm worried it'll get confused with all the closures.

Suggested change

return workflow.NewContinueAsNewError(ctx, WorkflowWithDC(d.dc), d.WorkerDeploymentWorkflowArgs)

return workflow.NewContinueAsNewError(ctx, WorkerDeploymentWorkflowType, d.WorkerDeploymentWorkflowArgs)

dnr · 2025-02-06T18:04:45Z

service/worker/workerdeployment/workflow.go

+	getMaxVersionsInDeployment := func(ctx workflow.Context) interface{} {
+		return dynamicconfig.MatchingMaxVersionsInDeployment.Get(d.dc)(d.a.namespace.Name().String())
+	}
+	intEq := func(a, b interface{}) bool {


same == here?

dnr · 2025-02-06T18:06:04Z

service/worker/workerdeployment/workflow.go

+		d.logger.Error("error decoding max versions: ", err)
+		return err
+	}
+	if len(d.State.Versions) >= maxVersions {


mixing the sideeffect and logic in one function feels a little weird, can you have one function to handle all the side effect business, returning a struct of concrete values (or a single one) and another function for the workflow logic (comparison)?

…v-limits # Conflicts: # service/worker/workerdeployment/workflow.go

carlydf · 2025-02-06T21:43:14Z

service/worker/workerdeployment/version_workflow.go

@@ -411,7 +411,9 @@ func (d *VersionWorkflowRunner) handleRegisterWorker(ctx workflow.Context, args
 	err = workflow.Await(ctx, func() bool { return d.VersionState.StartedDeploymentWorkflow })
 	if err != nil {
 		d.logger.Error("Update canceled before deployment workflow started")
-		return err
+		// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.


Suggested change

// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.

// TODO (Carly): This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.

carlydf · 2025-02-06T21:43:48Z

service/worker/workerdeployment/version_workflow.go

+		RequestId: d.newUUID(ctx),
+	}).Get(ctx, nil)
+	if err != nil {
+		// TODO: make sure the error message that goes to the user is informative and has the limit mentioned


Suggested change

// TODO: make sure the error message that goes to the user is informative and has the limit mentioned

// TODO (Carly): make sure the error message that goes to the user is informative and has the limit mentioned

carlydf · 2025-02-06T21:44:00Z

service/worker/workerdeployment/version_workflow.go

@@ -411,7 +411,9 @@ func (d *VersionWorkflowRunner) handleRegisterWorker(ctx workflow.Context, args
 	err = workflow.Await(ctx, func() bool { return d.VersionState.StartedDeploymentWorkflow })
 	if err != nil {
 		d.logger.Error("Update canceled before deployment workflow started")
-		return err
+		// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.
+		// TODO: mention the limit in here or make sure matching does in the error returned to the poller


Suggested change

// TODO: mention the limit in here or make sure matching does in the error returned to the poller

// TODO (Carly): mention the limit in here or make sure matching does in the error returned to the poller

carlydf · 2025-02-06T21:46:49Z

service/worker/workerdeployment/workflow.go

+	maxVersions := d.getMaxVersions(ctx)
+
+	if len(d.State.Versions) >= maxVersions {
+		return temporal.NewApplicationError(fmt.Sprintf("cannot add version, already at max versions %d", maxVersions), errTooManyVersions)


Suggested change

return temporal.NewApplicationError(fmt.Sprintf("cannot add version, already at max versions %d", maxVersions), errTooManyVersions)

err := d.tryDeleteVersion(ctx)

if err != nil {

return temporal.NewApplicationError(fmt.Sprintf("cannot add version, already at max versions %d", maxVersions), errTooManyVersions)

}

the test that tests this was skipped.. which is why it didn't catch this regfression

Oh, good catch!

carlydf

I approve this once d.tryDeleteVersion is added back. I can't technically approve it since it's my PR though

dnr

some suggestions and nitpicks but nothing blocking

dnr · 2025-02-06T21:43:28Z

service/worker/workerdeployment/drainage_workflow.go

+		refresh, err := getSafeDurationConfig(ctx, "getDrainageRefreshInterval", unsafeRefreshIntervalGetter, defaultVisibilityRefresh)
+		if err != nil {
+			return err
 		}
+		_ = workflow.Sleep(ctx, refresh)


the delay of a workflow.Sleep is allowed to change without determinism error, so actually you can use the config directly here. also ok to use sideeffect to consistent if you want.

dnr · 2025-02-06T21:45:38Z

service/worker/workerdeployment/util.go

@@ -139,3 +144,18 @@ func DecodeWorkerDeploymentMemo(memo *commonpb.Memo) *deploymentspb.WorkerDeploy
 	}
 	return &workerDeploymentWorkflowMemo
 }
+
+func getSafeDurationConfig(ctx workflow.Context, id string, unsafeGetter func() any, defaultValue time.Duration) (time.Duration, error) {


this can easily be generic if you want to do that now

have to do it later

dnr · 2025-02-06T21:46:37Z

service/worker/workerdeployment/version_workflow.go

@@ -411,7 +411,9 @@ func (d *VersionWorkflowRunner) handleRegisterWorker(ctx workflow.Context, args
 	err = workflow.Await(ctx, func() bool { return d.VersionState.StartedDeploymentWorkflow })
 	if err != nil {
 		d.logger.Error("Update canceled before deployment workflow started")
-		return err
+		// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.


Suggested change

// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.

// TODO: This is likely due to too many deployments, but make sure we excluded other possible errors here and send a proper error message all the time.

dnr · 2025-02-06T21:48:47Z

service/worker/workerdeployment/workflow.go

 	return nil
 }

+func (d *WorkflowRunner) getMaxVersions(ctx workflow.Context) int {


make getSafeDurationConfig generic and use it here?

…7268) ## What changed? Use MutableSideEffect to access dynamic config from inside workflows. ## Why? To prevent non-determinism in the workflow code. To allow the workflows to use the latest dynamic config values at all times. ## How did you test it? Tests are failing for other reasons, so I'm unable to test until I pull in those changes. ## Potential risks  ## Documentation  ## Is hotfix candidate?  --------- Co-authored-by: ShahabT <shahab.tajik@temporal.io>

carlydf requested a review from a team as a code owner February 6, 2025 04:15

carlydf changed the title ~~pass dynamic config collection to workflows~~ Use MutableSideEffect to access dynamic config from inside workflows Feb 6, 2025

ShahabT force-pushed the versioning-3.1-merge branch 3 times, most recently from 122d338 to 9140a68 Compare February 6, 2025 17:17

carlydf added 3 commits February 6, 2025 09:51

pass dynamic config collection to workflows

136c568

get dynamic config constants within workflow using MutableSideEffect

ed793e4

remove ramping backoff

e0fa16e

ShahabT force-pushed the cdf/wv-limits branch from 06ce272 to e0fa16e Compare February 6, 2025 17:53

dnr reviewed Feb 6, 2025

View reviewed changes

ShahabT and others added 5 commits February 6, 2025 12:19

Merge remote-tracking branch 'origin/versioning-3.1-merge' into cdf/w…

40f7004

…v-limits # Conflicts: # service/worker/workerdeployment/workflow.go

Add test and address feedback

5a7693f

touch ups

08dd7ca

add nits

9aa5de4

fix lint

7043208

carlydf commented Feb 6, 2025

View reviewed changes

dnr approved these changes Feb 6, 2025

View reviewed changes

address comments

3a8400a

ShahabT merged commit be8d90a into versioning-3.1-merge Feb 6, 2025
8 of 9 checks passed

ShahabT deleted the cdf/wv-limits branch February 6, 2025 21:58

	return workflow.NewContinueAsNewError(ctx, WorkflowWithDC(d.dc), d.WorkerDeploymentWorkflowArgs)
	return workflow.NewContinueAsNewError(ctx, WorkerDeploymentWorkflowType, d.WorkerDeploymentWorkflowArgs)

	// TODO: This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.
	// TODO (Carly): This is likely due to too many deployments, but make sure we exculed other possible errors here and send a proper error message all the time.

	// TODO: make sure the error message that goes to the user is informative and has the limit mentioned
	// TODO (Carly): make sure the error message that goes to the user is informative and has the limit mentioned

	// TODO: mention the limit in here or make sure matching does in the error returned to the poller
	// TODO (Carly): mention the limit in here or make sure matching does in the error returned to the poller

Use MutableSideEffect to access dynamic config from inside workflows #7268

Use MutableSideEffect to access dynamic config from inside workflows #7268

Uh oh!

Conversation

carlydf commented Feb 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changed?

Why?

How did you test it?

Potential risks

Documentation

Is hotfix candidate?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

carlydf left a comment

Choose a reason for hiding this comment

Uh oh!

dnr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

carlydf commented Feb 6, 2025 •

edited

Loading