Early-return sample

early-return/README.md

@@ -0,0 +1,17 @@
+### Early-Return Sample
+
+This sample demonstrates an early-return from a workflow.
+
+By utilizing Update-with-Start, a client can start a new workflow and synchronously receive 
+a response mid-workflow, while the workflow continues to run to completion.
+
+### Steps to run this sample:
+1) Run a [Temporal service](https://github.com/temporalio/samples-go/tree/main/#how-to-use).
+2) Run the following command to start the worker
+```
+go run early-return/worker/main.go
+```
+3) Run the following command to start the example
+```
+go run early-return/starter/main.go
+```

early-return/activity.go

@@ -0,0 +1,36 @@
+package earlyreturn
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"go.temporal.io/sdk/activity"
+)
+
+func InitTransaction(ctx context.Context, tx Transaction) error {
+	logger := activity.GetLogger(ctx)
+	if tx.FromAccount == "" {
+		return errors.New("invalid fromAccount")
+	}
+	if tx.ToAccount == "" {
+		return errors.New("invalid toAccount")
+	}
+	if tx.Amount == 0 {
+		return errors.New("invalid amount")
+	}
+	logger.Info("Transaction initialized")
+	return nil
+}
+
+func CancelTransaction(ctx context.Context, tx Transaction) {
+	logger := activity.GetLogger(ctx)
+	time.Sleep(1 * time.Second)
+	logger.Info("Transaction cancelled")
+}
+
+func CompleteTransaction(ctx context.Context, tx Transaction) {
+	logger := activity.GetLogger(ctx)
+	time.Sleep(1 * time.Second)
+	logger.Info("Transaction completed")
+}

early-return/starter/main.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/pborman/uuid"
+	"github.com/temporalio/samples-go/early-return"
+	"go.temporal.io/sdk/client"
+)
+
+func main() {
+	c, err := client.Dial(client.Options{})
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	ctxWithTimeout, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	updateOperation := client.NewUpdateWithStartWorkflowOperation(
+		client.UpdateWorkflowOptions{
+			UpdateName:   earlyreturn.UpdateName,
+			WaitForStage: client.WorkflowUpdateStageCompleted,
+		})
+
+	tx := earlyreturn.Transaction{ID: uuid.New(), FromAccount: "Bob", ToAccount: "Alice", Amount: 100.0}
+	workflowOptions := client.StartWorkflowOptions{
+		ID:                 "early-return-workflow-ID-" + tx.ID,
+		TaskQueue:          earlyreturn.TaskQueueName,
+		WithStartOperation: updateOperation,
+	}
+	we, err := c.ExecuteWorkflow(ctxWithTimeout, workflowOptions, earlyreturn.Workflow, tx)
+	if err != nil {
+		log.Fatalln("Error executing workflow:", err)
+	}
+
+	log.Println("Started workflow", "WorkflowID", we.GetID(), "RunID", we.GetRunID())
+
+	updateHandle, err := updateOperation.Get(ctxWithTimeout)
+	if err != nil {
+		log.Fatalln("Error obtaining update handle:", err)
+	}
+
+	err = updateHandle.Get(ctxWithTimeout, nil)
+	if err != nil {
+		// NOTE: If the error is retryable, a retry attempt must use a unique workflow ID.
+		log.Fatalln("Error obtaining update result:", err)
+	}
+
+	log.Println("Transaction initialized successfully")
+
+	// The workflow will continue running, either completing or cancelling the transaction.
+}

early-return/worker/main.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"log"
+
+	"github.com/temporalio/samples-go/early-return"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+)
+
+func main() {
+	// The client and worker are heavyweight objects that should be created once per process.
+	c, err := client.Dial(client.Options{})
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	w := worker.New(c, earlyreturn.TaskQueueName, worker.Options{})
+
+	w.RegisterWorkflow(earlyreturn.Workflow)
+
+	err = w.Run(worker.InterruptCh())
+	if err != nil {
+		log.Fatalln("Unable to start worker", err)
+	}
+}

early-return/workflow.go

@@ -0,0 +1,82 @@
+package earlyreturn
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"go.temporal.io/sdk/workflow"
+)
+
+const (
+	UpdateName    = "early-return"
+	TaskQueueName = "early-return-tq"
+)
+
+var (
+	activityTimeout    = 2 * time.Second
+	earlyReturnTimeout = 5 * time.Second
+)
+
+type Transaction struct {
+	ID          string
+	FromAccount string
+	ToAccount   string
+	Amount      float64
+}
+
+// Workflow processes a transaction in two phases. First, the transaction is initialized, and if successful,
+// it proceeds to completion. However, if initialization fails - due to validation errors or transient
+// issues (e.g., network connectivity problems) - the transaction is cancelled.
+//
+// By utilizing Update-with-Start, the client can initiate the workflow and immediately receive the result of
+// the initialization in a single round trip, even before the transaction processing completes. The remainder
+// of the transaction is then processed asynchronously.
+func Workflow(ctx workflow.Context, tx Transaction) error {
+	var initErr error
+	var initDone bool
+	logger := workflow.GetLogger(ctx)
+
+	if err := workflow.SetUpdateHandler(
+		ctx,
+		UpdateName,
+		func(ctx workflow.Context) error {
+			condition := func() bool { return initDone }
+			if completed, err := workflow.AwaitWithTimeout(ctx, earlyReturnTimeout, condition); err != nil {
+				return fmt.Errorf("update cancelled: %w", err)
+			} else if !completed {
+				return errors.New("update timed out")
+			}
+			return initErr
+		},
+	); err != nil {
+		return err
+	}
+
+	// Phase 1: Initialize the transaction synchronously.
+	//
+	// By using a local activity, an additional server roundtrip is avoided.
+	// See https://docs.temporal.io/activities#local-activity for more details.
+
+	activityOptions := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+		ScheduleToCloseTimeout: activityTimeout,
+	})
+	initErr = workflow.ExecuteLocalActivity(activityOptions, InitTransaction, tx).Get(ctx, nil)
+	initDone = true
+
+	// Phase 2: Complete or cancel the transaction asychronously.
+	activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+		StartToCloseTimeout: 10 * time.Second,
+	})
+	if initErr != nil {
+		logger.Info("cancelling transaction due to error: %v", initErr)
+
+		// Transaction failed to be initialized or not quickly enough; cancel the transaction.
+		return workflow.ExecuteActivity(activityCtx, CancelTransaction, tx).Get(ctx, nil)
+	}
+
+	logger.Info("completing transaction")
+
+	// Transaction was initialized successfully; complete the transaction.
+	return workflow.ExecuteActivity(activityCtx, CompleteTransaction, tx).Get(ctx, nil)
+}