Add command for workflow update

common/defs-cmds.go

@@ -25,6 +25,7 @@ const (
 	DeleteWorkflowDefinition    = "Deletes a Workflow Execution."
 	ResetWorkflowDefinition     = "Resets a Workflow Execution by Event ID or reset type."
 	TraceWorkflowDefinition     = "Trace progress of a Workflow Execution and its children."
+	UpdateWorkflowDefinition    = "Updates a running workflow synchronously."
 
 	// Activity subcommand definitions
 	CompleteActivityDefinition = "Completes an Activity Execution."
@@ -197,7 +198,6 @@ Workflow terminations require a valid [Workflow ID](/concepts/what-is-a-workflow
 
 Use the options listed below to change termination behavior.`
 
-
 const ResetWorkflowUsageText = `The ` + "`" + `temporal workflow reset` + "`" + ` command resets a [Workflow Execution](/concepts/what-is-a-workflow-execution).
 A reset allows the Workflow to resume from a certain point without losing its parameters or [Event History](/concepts/what-is-an-event-history).
 
@@ -547,6 +547,10 @@ const WorkflowTraceUsageText = `The ` + "`" + `temporal workflow trace` + "`" +
 
 Use the options listed below to change the command's behavior.`
 
+const WorkflowUpdateUsageText = `The ` + "`" + `temporal workflow update` + "`" + ` command synchronously updates a running [Workflow Execution](/concepts/what-is-a-workflow-execution).
+
+Use the options listed below to change the command's behavior.`
+
 const ServerUsageText = `Server commands allow you to start and manage the [Temporal Server](/concepts/what-is-the-temporal-server) from the command line.
 
 Currently, ` + "`" + `cli` + "`" + ` server functionality extends to starting the Server. 
@@ -571,4 +575,4 @@ USAGE:
 
 DISPLAY OPTIONS:
    {{template "visibleFlagTemplate" .}}{{end}}
-`
+`

common/flags.go

@@ -118,6 +118,9 @@ var (
 	FlagUICodecEndpoint            = "ui-codec-endpoint"
 	FlagUIPort                     = "ui-port"
 	FlagUnpause                    = "unpause"
+	FlagUpdateWaitPolicy           = "wait-policy"
+	FlagUpdateFirstExecutionRunID  = "first-execution-run-id"
+	FlagUpdateID                   = "update-id"
 	FlagVisibilityArchivalState    = "visibility-archival-state"
 	FlagVisibilityArchivalURI      = "visibility-uri"
 	FlagWorkflowExecutionTimeout   = "execution-timeout"

tests/e2e_test.go

@@ -35,12 +35,22 @@ func TestClientIntegrationSuite(t *testing.T) {
 
 func (s *e2eSuite) SetupSuite() {
 	s.app = app.BuildApp()
-	server, err := testsuite.StartDevServer(context.Background(), testsuite.DevServerOptions{})
+	server, err := testsuite.StartDevServer(context.Background(), testsuite.DevServerOptions{
+		ExtraArgs: []string{
+			// server logs are too noisy, limit server logs
+			"--log-level", "error",
+			//TODO: remove this flag when update workflow is enabled in the server by default
+			"--dynamic-config-value", "frontend.enableUpdateWorkflowExecution=true",
+		},
+	})
 	s.NoError(err)
 	s.ts = server
 }
 
 func (s *e2eSuite) TearDownSuite() {
+	err := s.ts.Stop()
+	s.NoError(err)
+
 }
 
 func (s *e2eSuite) SetupTest() {
@@ -51,10 +61,12 @@ func (s *e2eSuite) SetupTest() {
 	})
 	s.writer = &MemWriter{}
 	s.app.Writer = s.writer
+
+	// noop exiter to prevent the app from exiting mid test
+	cli.OsExiter = func(code int) { return }
 }
 
 func (s *e2eSuite) TearDownTest() {
-	s.ts.Stop()
 }
 
 func (s *e2eSuite) NewWorker(taskQueue string, registerFunc func(registry worker.Registry)) worker.Worker {

tests/workflow_test.go

@@ -2,12 +2,16 @@ package tests
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"os"
+	"strconv"
 
 	"github.com/pborman/uuid"
 	"github.com/temporalio/cli/tests/workflows/helloworld"
-	"go.temporal.io/sdk/client"
+	"github.com/temporalio/cli/tests/workflows/update"
+	sdkclient "go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/worker"
 )
 
@@ -25,7 +29,7 @@ func (s *e2eSuite) TestWorkflowShow_ReplayableHistory() {
 
 	wfr, err := c.ExecuteWorkflow(
 		context.Background(),
-		client.StartWorkflowOptions{TaskQueue: testTq},
+		sdkclient.StartWorkflowOptions{TaskQueue: testTq},
 		helloworld.Workflow,
 		"world",
 	)
@@ -51,3 +55,52 @@ func (s *e2eSuite) TestWorkflowShow_ReplayableHistory() {
 	err = replayer.ReplayWorkflowHistoryFromJSONFile(nil, historyFile)
 	s.NoError(err)
 }
+
+func (s *e2eSuite) TestWorkflowUpdate() {
+	c := s.ts.Client()
+
+	s.NewWorker(testTq, func(r worker.Registry) {
+		r.RegisterWorkflow(update.Counter)
+	})
+	randomInt := rand.Intn(100)
+	wfr, err := c.ExecuteWorkflow(
+		context.Background(),
+		sdkclient.StartWorkflowOptions{TaskQueue: testTq},
+		update.Counter,
+		randomInt,
+	)
+	s.NoError(err)
+	signalWorkflow := func() {
+		// send a Signal to stop the workflow
+		err = c.SignalWorkflow(context.Background(), wfr.GetID(), wfr.GetRunID(), update.Done, nil)
+		s.NoError(err)
+	}
+
+	defer signalWorkflow()
+
+	// successful update with wait policy Completed, should show the result
+	err = s.app.Run([]string{"", "workflow", "update", "--workflow-id", wfr.GetID(), "--run-id", wfr.GetRunID(), "--name", update.FetchAndAdd, "-i", strconv.Itoa(randomInt)})
+	s.NoError(err)
+	want := fmt.Sprintf(": %v", randomInt)
+	s.Contains(s.writer.GetContent(), want)
+
+	// successful update with wait policy Completed, passing first-execution-run-id
+	err = s.app.Run([]string{"", "workflow", "update", "--workflow-id", wfr.GetID(), "--run-id", wfr.GetRunID(), "--name", update.FetchAndAdd, "-i", "1", "--first-execution-run-id", wfr.GetRunID()})
+	s.NoError(err)
+
+	// update rejected, when name is not available
+	err = s.app.Run([]string{"", "workflow", "update", "--workflow-id", "non-existent-ID", "--run-id", wfr.GetRunID(), "-i", "1"})
+	s.Error(err)
+	s.ErrorContains(err, "Required flag \"name\" not set")
+
+	// update rejected, wrong workflowID
+	err = s.app.Run([]string{"", "workflow", "update", "--workflow-id", "non-existent-ID", "--run-id", wfr.GetRunID(), "--name", update.FetchAndAdd, "-i", "1"})
+	s.Error(err)
+	s.ErrorContains(err, "update workflow failed")
+
+	// update rejected, wrong update name
+	err = s.app.Run([]string{"", "workflow", "update", "--workflow-id", wfr.GetID(), "--run-id", wfr.GetRunID(), "--name", "non-existent-name", "-i", "1"})
+	s.Error(err)
+	s.ErrorContains(err, "update workflow failed: unknown update")
+
+}

tests/workflows/update/update.go

@@ -0,0 +1,44 @@
+package update
+
+import (
+	"fmt"
+
+	"go.temporal.io/sdk/workflow"
+)
+
+const (
+	FetchAndAdd = "fetch_and_add"
+	Done        = "done"
+)
+
+func Counter(ctx workflow.Context, val int) (int, error) {
+	log := workflow.GetLogger(ctx)
+	counter := val
+
+	if err := workflow.SetUpdateHandlerWithOptions(
+		ctx,
+		FetchAndAdd,
+		func(ctx workflow.Context, i int) (int, error) {
+			tmp := counter
+			counter += i
+			log.Info("counter updated", "added", i, "new-value", counter)
+			return tmp, nil
+		},
+		workflow.UpdateHandlerOptions{Validator: nonNegative},
+	); err != nil {
+		return 0, err
+	}
+
+	_ = workflow.GetSignalChannel(ctx, Done).Receive(ctx, nil)
+	return counter, ctx.Err()
+}
+
+func nonNegative(ctx workflow.Context, i int) error {
+	log := workflow.GetLogger(ctx)
+	if i < 0 {
+		log.Debug("Rejecting negative update", "addend", i)
+		return fmt.Errorf("addend must be non-negative (%v)", i)
+	}
+	log.Debug("Accepting update", "addend", i)
+	return nil
+}