NEXUS-474: Add sample for Workflow Update Nexus op

go.mod

@@ -1,9 +1,11 @@
 module github.com/temporalio/samples-go
 
-go 1.25.0
+go 1.25.4
 
 replace github.com/cactus/go-statsd-client => github.com/cactus/go-statsd-client/v5 v5.0.0
 
+replace go.temporal.io/sdk => ../sdk-go
+
 require (
 	github.com/aws/aws-sdk-go-v2 v1.41.7
 	github.com/aws/aws-sdk-go-v2/config v1.32.17
@@ -24,7 +26,7 @@ require (
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.16.0
 	go.opentelemetry.io/otel/sdk v1.42.0
 	go.opentelemetry.io/otel/trace v1.42.0
-	go.temporal.io/api v1.62.12
+	go.temporal.io/api v1.62.14
 	go.temporal.io/sdk v1.44.0
 	go.temporal.io/sdk/contrib/aws/lambdaworker v0.1.1
 	go.temporal.io/sdk/contrib/aws/lambdaworker/otel v0.1.1
@@ -109,6 +111,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect
 	github.com/outcaste-io/ristretto v0.2.3 // indirect
 	github.com/philhofer/fwd v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
@@ -159,7 +162,7 @@ require (
 	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
 	golang.org/x/tools v0.41.0 // indirect
-	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
+	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect

nexus-operations/update-workflow/README.md

@@ -0,0 +1,123 @@
+# Nexus Operations: UpdateWorkflow 
+
+This sample shows how to expose and run an `UpdateWorkflow` Temporal operation as a Nexus operation. The operation sends an update to a running workflow - so it is required that the workflowID be known and that the worklow isn't yet in a terminal state just as is the case for a normal `UpdateWorkflow` operation
+
+For Nexus Operation wrapped `UpdateWorkflow` operations, only `WorkflowUpdateStageAccepted` is accepted as the `WaitForStage`. This is because nexus ops need to complete within 10s and `WorkflowUpdateStageCompleted` will wait until the update is fully finished - this could lead to cases where Updates keep getting retried on a consistently slow operation
+
+## Running the samples
+
+There are two samples in this repo 
+1. Nexus Op UpdateWorkfow inside in a workflow. This is for cases where the Nexus Op will be called inside a workflow - see `caller` folder for details
+2. Standalone Nexus Op Update Workflow. This is for use-cases where the Nexus Op needs to be called from outside of a workflow. See the `standalone` folder for details 
+
+For both cases, the handler remains the same 
+
+### Running Standalone Nexus Operation UpdateWorkflow
+
+Ensure the temporal server has the required configurations. If required, start the dev server locally with the dynamic config flags required for standalone Nexus operations:
+
+TODO: add/estimate version of the temporal server that has the SANO feature(behind gates, etc)
+
+```
+temporal server start-dev \
+  --dynamic-config-value "nexusoperation.enableStandalone=true" \
+  --dynamic-config-value "history.enableCHASMCallbacks=true" \
+  --dynamic-config-value "history.enableUpdateCallbacks=true"
+```
+
+#### Create the handler namespace
+
+```
+temporal operator namespace create --namespace my-target-namespace
+```
+
+#### Create the Nexus endpoint
+
+```
+temporal operator nexus endpoint create \
+  --name counter-update-endpoint \
+  --target-namespace my-target-namespace \
+  --target-task-queue counter-update-handler-tq
+```
+
+#### Start the Handler worker 
+
+```
+go run ./nexus-operations/update-workflow/handler/worker -namespace my-target-namespace
+```
+
+#### Start the Handler workflow (receiver)
+
+Start the handler workflow so that it can receive the updates 
+
+```
+go run ./nexus-operations/update-workflow/handler/starter -namespace my-target-namespace
+```
+
+NOTE:
+Send a `done` signal to the counter workflow to close it  
+
+```
+temporal workflow signal --namespace my-target-namespace --workflow-id counter-workflow-1 --name done
+```
+
+#### Run the Standalone Nexus Operation 
+
+```
+go run ./nexus-operations/update-workflow/standalone -namespace my-target-namespace -incr 1
+```
+
+### Running Nexus Operation UpdateWorkflow inside a Workflow
+
+#### Create the caller and handler namespaces
+
+```
+temporal operator namespace create --namespace my-target-namespace
+temporal operator namespace create --namespace my-caller-namespace
+```
+
+#### Create the Nexus Endpoint
+
+```
+temporal operator nexus endpoint create \
+  --name counter-update-endpoint \
+  --target-namespace my-target-namespace \
+  --target-task-queue counter-update-handler-tq
+```
+
+#### Setting up the Workers 
+
+##### Handler worker (target namespace)
+
+```
+go run ./nexus-operations/update-workflow/handler/worker -namespace my-target-namespace
+```
+
+##### Caller worker (caller namespace)
+
+```
+go run ./nexus-operations/update-workflow/caller/worker -namespace my-caller-namespace
+```
+
+#### Set up the receiver 
+
+Start the handler workflow so that it can receive the updates 
+
+```
+go run ./nexus-operations/update-workflow/handler/starter -namespace my-target-namespace
+```
+
+NOTE:
+Send a `done` signal to the counter workflow to close it  
+
+```
+temporal workflow signal --namespace my-target-namespace --workflow-id counter-workflow-1 --name done
+```
+
+#### Trigger the Nexus UpdateWorkflow operation
+
+```
+go run ./nexus-operations/update-workflow/caller/starter -namespace my-caller-namespace 
+```
+
+

nexus-operations/update-workflow/api/api.go

@@ -0,0 +1,27 @@
+package api
+
+const (
+	// Nexus service exposing the Update operation
+	CounterUpdateServiceName = "counter-update-service"
+	// Name of the Nexus Operation- backed by the UpdateWorkflow- to bump counter
+	IncrOperationName = "incr"
+	// Name of the Update receiver on the handler
+	IncrUpdateName = "incr"
+	// Signal for the long running counter workflow(handler) to complete
+	DoneSignalName = "done"
+	// Task queue for the handler worker to process counter updates
+	HandlerTaskQueueName = "counter-update-handler-tq"
+	// Workflow ID of the handler workflow. Required to be known by the caller
+	CounterWorkflowID = "counter-workflow-1"
+	// Nexus endpoint
+	EndpointName = "counter-update-endpoint"
+)
+
+type Input struct {
+	WorkflowID string
+	Incr       int
+}
+
+type Output struct {
+	NewCount int
+}

nexus-operations/update-workflow/caller/starter/main.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"time"
+
+	"go.temporal.io/sdk/client"
+
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/api"
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/caller"
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/options"
+)
+
+func main() {
+	set := flag.NewFlagSet("nexus-update-op-caller-starter", flag.ExitOnError)
+	fp := options.NewClientFlagParser(set)
+	incrementAmount := set.Int("incr", 1, "increment amount, defaults to 1 if <= 0")
+	set.Parse(os.Args[1:])
+	clientOptions, err := fp.ClientOptions()
+	if err != nil {
+		log.Fatalf("Invalid options: %v", err)
+	}
+	c, err := client.Dial(clientOptions)
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	ctx := context.Background()
+	workflowOptions := client.StartWorkflowOptions{
+		ID:        "counter-update-caller-" + time.Now().Format("20060102150405"),
+		TaskQueue: caller.TaskQueue,
+	}
+	input := api.Input{WorkflowID: api.CounterWorkflowID, Incr: *incrementAmount}
+
+	log.Println("Invoking incr operation")
+	wr, err := c.ExecuteWorkflow(ctx, workflowOptions, caller.UpdateRemoteCounterWorkflow, input)
+	if err != nil {
+		log.Fatalln("Unable to execute workflow", err)
+	}
+	log.Println("Started workflow", "WorkflowID", wr.GetID(), "RunID", wr.GetRunID())
+
+	var out api.Output
+	if err := wr.Get(ctx, &out); err != nil {
+		log.Fatalln("Unable to get workflow result", err)
+	}
+	log.Println("Counter new value", out.NewCount)
+}

nexus-operations/update-workflow/caller/worker/main.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/caller"
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/options"
+)
+
+func main() {
+	set := flag.NewFlagSet("nexus-update-op-caller-worker", flag.ExitOnError)
+	fp := options.NewClientFlagParser(set)
+	set.Parse(os.Args[1:])
+	clientOptions, err := fp.ClientOptions()
+	if err != nil {
+		log.Fatalf("Invalid options: %v", err)
+	}
+	c, err := client.Dial(clientOptions)
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	w := worker.New(c, caller.TaskQueue, worker.Options{})
+	w.RegisterWorkflow(caller.UpdateRemoteCounterWorkflow)
+
+	if err := w.Run(worker.InterruptCh()); err != nil {
+		log.Fatalln("Unable to start worker", err)
+	}
+}

nexus-operations/update-workflow/caller/workflow.go

@@ -0,0 +1,28 @@
+package caller
+
+import (
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/api"
+	"go.temporal.io/sdk/workflow"
+)
+
+const (
+	TaskQueue = "remote-update-caller-tq"
+)
+
+// Executes the UpdateWorkflow on the handler namespace's workflow via Nexus Operation
+func UpdateRemoteCounterWorkflow(ctx workflow.Context, input api.Input) (api.Output, error) {
+	c := workflow.NewNexusClient(api.EndpointName, api.CounterUpdateServiceName)
+
+	fut := c.ExecuteOperation(ctx, api.IncrOperationName, input, workflow.NexusOperationOptions{})
+
+	var exec workflow.NexusOperationExecution // exec token can be used to cancel if the update supports it
+	if err := fut.GetNexusOperationExecution().Get(ctx, &exec); err != nil {
+		return api.Output{}, err
+	}
+
+	var out api.Output
+	if err := fut.Get(ctx, &out); err != nil {
+		return api.Output{}, err
+	}
+	return out, nil
+}

nexus-operations/update-workflow/handler/app.go

@@ -0,0 +1,69 @@
+package handler
+
+import (
+	"context"
+	"math/rand"
+	"time"
+
+	"github.com/temporalio/samples-go/nexus-operations/update-workflow/api"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporalnexus"
+	"go.temporal.io/sdk/workflow"
+)
+
+// IncrOperation exposes the Workflow Update as a Nexus operation(via StartUpdateWorkflow)
+var IncrOperation = temporalnexus.MustNewTemporalOperation(
+	temporalnexus.TemporalOperationOptions[api.Input, api.Output]{
+		Name: api.IncrOperationName,
+		Start: func(
+			ctx context.Context,
+			nc temporalnexus.NexusClient,
+			input api.Input,
+			options temporalnexus.StartTemporalOperationOptions,
+		) (temporalnexus.TemporalOperationResult[api.Output], error) {
+			return temporalnexus.StartUpdateWorkflow[api.Output](ctx, nc, client.UpdateWorkflowOptions{
+				WorkflowID:   input.WorkflowID,
+				UpdateName:   api.IncrUpdateName,
+				Args:         []any{input.Incr},
+				WaitForStage: client.WorkflowUpdateStageAccepted,
+			})
+		},
+	},
+)
+
+// CounterWorkflow is a long-running workflow that handles incr requests and stays open until it
+// receives the Done signal.
+func CounterWorkflow(ctx workflow.Context) (int, error) {
+	logger := workflow.GetLogger(ctx)
+	count := 0
+
+	if err := workflow.SetUpdateHandler(
+		ctx,
+		api.IncrUpdateName,
+		func(ctx workflow.Context, incr int) (api.Output, error) {
+			if incr <= 0 {
+				incr = 1
+			}
+			count += incr
+			newCount := count
+			logger.Info("counter updated", "incr", incr, "newValue", newCount)
+			var randomSeconds int
+			if err := workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} {
+				return rand.Intn(6) // sleep upto 5 seconds
+			}).Get(&randomSeconds); err != nil {
+				logger.Error("unexpected error", err)
+				return api.Output{}, err
+			}
+			if err := workflow.Sleep(ctx, time.Duration(randomSeconds)*time.Second); err != nil {
+				logger.Error("unexpected error", err)
+				return api.Output{}, err
+			}
+			return api.Output{NewCount: newCount}, nil
+		},
+	); err != nil {
+		return 0, err
+	}
+
+	workflow.GetSignalChannel(ctx, api.DoneSignalName).Receive(ctx, nil)
+	return count, nil
+}