design-patterns/go

70 notes in this subject, read out of this brain and free to use. This is page 1 of 2.

Go Activity Dependency Injection implementation

In Go, Activities are methods on a struct with fields holding dependencies. The struct is registered with the Worker at startup. Workflows reference Activities using a nil pointer of the Activity struct type (e.g., `var a *Activities`) which provides compile-time method references without instantiating the struct. The Temporal runtime resolves the actual registered instance at execution time.

Go circuit breaker example with dependency injection

// activities.go — github.com/sony/gobreaker/v2 package payment import ( "context" "github.com/sony/gobreaker/v2" "go.temporal.io/sdk/activity" ) type Activities struct { PaymentAPI PaymentAPI Breaker *gobreaker.CircuitBreaker[string] } func (a *Activities) ChargeCustomer(ctx context.Context, orderID string, amount int) (string, error) { logger := activity.GetLogger(ctx) logger.Info("Charging customer", "orderID", orderID, "amount", amount) // The breaker rejects the call immediately when it is open, // returning gobreaker.ErrOpenState without touching the API. return a.Breaker.Execute(func() (string, error) { return a.PaymentAPI.Charge(orderID, amount) }) } // worker/main.go package main import ( "log" "time" "github.com/sony/gobreaker/v2" "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" "example/payment" ) func main() { c, err := client.Dial(client.Options{}) if err != nil { log.Fatalln("Unable to create client", err) } defer c.Close() w := worker.New(c, "payment", worker.Options{}) w.RegisterWorkflow(payment.PaymentWorkflow) // Construct the breaker once at Worker startup so its failure // counters are shared across all Activity executions. breaker := gobreaker.NewCircuitBreaker[string](gobreaker.Settings{ Name: "payment-api", MaxRequests: 3, // probes allowed while half-open Interval: 60 * time.Second, // window for counting failures Timeout: 30 * time.Second, // cool-down before half-open }) w.RegisterActivity(&payment.Activities{ PaymentAPI: payment.NewPaymentAPI("https://api.example.com"), Breaker: breaker, }) err = w.Run(worker.InterruptCh()) if err != nil { log.Fatalln("Unable to start worker", err) } }

Go Activity Dependency Injection code example

// activities.go package payment import ( "context" "go.temporal.io/sdk/activity" ) type Activities struct { DBClient DBClient EmailClient EmailClient } func (a *Activities) ChargeCustomer(ctx context.Context, orderID string, amount int) (string, error) { logger := activity.GetLogger(ctx) logger.Info("Charging customer", "orderID", orderID, "amount", amount) receiptID, err := a.DBClient.ProcessPayment(orderID, amount) if err != nil { return "", err } return receiptID, nil } func (a *Activities) SendReceipt(ctx context.Context, email string, receiptID string) error { return a.EmailClient.Send(email, "Payment Receipt", receiptID) } // worker/main.go package main import ( "log" "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" "example/payment" ) func main() { c, err := client.Dial(client.Options{}) if err != nil { log.Fatalln("Unable to create client", err) } defer c.Close() w := worker.New(c, "payment", worker.Options{}) w.RegisterWorkflow(payment.PaymentWorkflow) // Inject real dependencies at Worker startup w.RegisterActivity(&payment.Activities{ DBClient: payment.NewPostgresClient("postgres://localhost:5432/payments"), EmailClient: payment.NewSMTPClient("smtp://mail.example.com"), }) err = w.Run(worker.InterruptCh()) if err != nil { log.Fatalln("Unable to start worker", err) } } // workflow.go package payment import ( "time" "go.temporal.io/sdk/workflow" ) func PaymentWorkflow(ctx workflow.Context, orderID string, amount int, email string) error { ao := workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) // Use a nil struct pointer to reference Activity methods. // This provides compile-time type safety without instantiating the struct. var a *Activities var receiptID string err := workflow.ExecuteActivity(ctx, a.ChargeCustomer, orderID, amount).Get(ctx, &receiptID) if err != nil { return err } return workflow.ExecuteActivity(ctx, a.SendReceipt, email, receiptID).Get(ctx, nil) }

Batch Iterator Go implementation

package main import ( "time" "go.temporal.io/sdk/workflow" ) func BatchIteratorWorkflow(ctx workflow.Context, offset int, totalProcessed int) (int, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) var page []Record if err := workflow.ExecuteActivity(ctx, FetchPage, offset, PageSize).Get(ctx, &page); err != nil { return totalProcessed, err } for _, record := range page { if err := workflow.ExecuteActivity(ctx, ProcessRecord, record).Get(ctx, nil); err != nil { return totalProcessed, err } totalProcessed++ } workflow.GetLogger(ctx).Info("Processed page", "offset", offset, "pageSize", len(page), "totalProcessed", totalProcessed) if len(page) == PageSize { return totalProcessed, workflow.NewContinueAsNewError(ctx, BatchIteratorWorkflow, offset+PageSize, totalProcessed) } return totalProcessed, nil }

Parallel child workflows in Go

To start multiple child workflows in parallel in Go, loop through items calling workflow.ExecuteChildWorkflow(ctx, ChildWorkflow, item) which returns immediately without blocking, storing the futures. Then iterate through the futures calling .Get(ctx, &result) on each to collect results.

Synchronous child workflow execution in Go

To execute a child workflow synchronously in Go, call workflow.ExecuteChildWorkflow(ctx, ChildWorkflow, input) which returns a ChildWorkflowFuture, then call .Get(ctx, &result) to block until the child completes.

Asynchronous child workflow execution in Go

To start a child workflow asynchronously in Go, call workflow.ExecuteChildWorkflow(ctx, ChildWorkflow, input) with a ChildWorkflowOptions set to ParentClosePolicy.ABANDON. Call childFuture.GetChildWorkflowExecution().Get(ctx, &childWE) to get the child's execution info without waiting for completion.

Go Continue-As-New implementation

In Go, call `workflow.NewContinueAsNewError(ctx, DataProcessorWorkflow, cursor, totalProcessed)` and return the error from the Workflow function. This signals the runtime to continue as new. Use `workflow.GetInfo(ctx).GetContinueAsNewSuggested()` to check if history is approaching the limit.

Delayed Start Go API

In Go, use `client.ExecuteWorkflow()` with `StartWorkflowOptions` containing a `StartDelay` field set to a `time.Duration`: ```go workflowOptions := client.StartWorkflowOptions{ ID: WorkflowID, TaskQueue: TaskQueue, StartDelay: 30 * time.Second, } we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, DelayedStartWorkflow) ```

Go workflow routing to rate-limited queue

Example of a Go Workflow routing Activities to a rate-limited queue: ```go // workflow.go func MyWorkflow(ctx workflow.Context, input string) (string, error) { ao := workflow.ActivityOptions{ TaskQueue: "rate-limited-tq", StartToCloseTimeout: 30 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) var result string err := workflow.ExecuteActivity(ctx, CallApi, input).Get(ctx, &result) return result, err } ``` The Workflow specifies an explicit task_queue override in the Activity options to route the throttled Activity to the dedicated queue.

Go worker with rate limiting

Example of a Go Worker configured with downstream rate limiting: ```go // main.go w := worker.New(c, "rate-limited-tq", worker.Options{ TaskQueueActivitiesPerSecond: 5.0, }) w.RegisterActivity(CallApi) if err := w.Run(worker.InterruptCh()); err != nil { log.Fatalf("worker error: %v", err) } ``` This worker is dedicated to rate-limited activities and requires a separate worker registered on the workflow task queue.

Go activity for downstream rate limiting

Example of a simple Go Activity for downstream rate limiting: ```go // activities.go func CallApi(ctx context.Context, input string) (string, error) { return downstreamApi.Call(input) } ```

Early Return Go implementation example

Example showing Early Return pattern in Go using Update-with-Start: ```go // workflow.go func Workflow(ctx workflow.Context, txRequest TransactionRequest) (*Transaction, error) { var tx *Transaction var initDone bool var initErr error // Register update handler that waits for initialization workflow.SetUpdateHandler(ctx, UpdateName, func(ctx workflow.Context) (*Transaction, error) { workflow.Await(ctx, func() bool { return initDone }) return tx, initErr }, ) // Phase 1: Fast synchronous initialization (local activity) localOpts := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{ ScheduleToCloseTimeout: 5 * time.Second, }) initErr = workflow.ExecuteLocalActivity(localOpts, txRequest.Init).Get(ctx, &tx) initDone = true // Signal update handler // Phase 2: Slow asynchronous completion activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, }) if initErr != nil { // Cancel on initialization failure return nil, workflow.ExecuteActivity(activityCtx, CancelTransaction, tx).Get(ctx, nil) } // Complete on initialization success return tx, workflow.ExecuteActivity(activityCtx, CompleteTransaction, tx).Get(ctx, nil) } // client.go startOp := client.NewWithStartWorkflowOperation( client.StartWorkflowOptions{ ID: "transaction-123", TaskQueue: "transactions", WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_FAIL, }, Workflow, txRequest, ) updateHandle, err := client.UpdateWithStartWorkflow(ctx, client.UpdateWithStartWorkflowOptions{ StartWorkflowOperation: startOp, UpdateOptions: client.UpdateWorkflowOptions{ UpdateName: UpdateName, WaitForStage: client.WorkflowUpdateStageCompleted, }, }, ) // Get initialization result immediately var tx Transaction err = updateHandle.Get(ctx, &tx) if err != nil { return err } // Use transaction ID immediately while workflow continues fmt.Printf("Transaction initialized: %s\n", tx.ID) ```

Entity Workflow Go implementation

type UserAccountWorkflow struct{} type UserState struct { Status string Profile ProfileData PendingEmail string CreatedAt time.Time UpdatedAt time.Time } type UserAccountInput struct { UserID string State *UserState } func (w *UserAccountWorkflow) Run(ctx workflow.Context, input UserAccountInput) error { var state UserState if input.State != nil { state = *input.State } else { state = UserState{ Status: "ACTIVE", CreatedAt: workflow.Now(ctx), } } deleted := false operationCount := 0 err := workflow.SetUpdateHandler(ctx, "updateProfile", func(ctx workflow.Context, data ProfileData) error { if deleted { return errors.New("user account is deleted") } if err := workflow.ExecuteActivity(ctx, ValidateProfile, data).Get(ctx, nil); err != nil { return err } state.Profile = data state.UpdatedAt = workflow.Now(ctx) operationCount++ return nil }) if err != nil { return err } err = workflow.SetUpdateHandler(ctx, "suspend", func(ctx workflow.Context) error { if !deleted && state.Status != "SUSPENDED" { state.Status = "SUSPENDED" state.UpdatedAt = workflow.Now(ctx) operationCount++ } return nil }) if err != nil { return err } for { selector := workflow.NewSelector(ctx) selector.AddReceive(workflow.GetSignalChannel(ctx, "delete"), func(c workflow.ReceiveChannel, more bool) { c.Receive(ctx, nil) deleted = true }) selector.Select(ctx) if deleted { state.Status = "DELETED" return nil } if workflow.GetInfo(ctx).GetContinueAsNewSuggested() { return workflow.NewContinueAsNewError(ctx, w.Run, UserAccountInput{UserID: input.UserID, State: &state}) } } }

Fan-Out Go implementation

```go // workflows.go package main import ( "fmt" "time" "go.temporal.io/sdk/workflow" ) func FanOutWorkflow(ctx workflow.Context, totalRecords int, chunkSize int) (int, error) { if chunkSize <= 0 { chunkSize = ChunkSize } var futures []workflow.Future parentID := workflow.GetInfo(ctx).WorkflowExecution.ID for offset := 0; offset < totalRecords; offset += chunkSize { length := chunkSize if offset+chunkSize > totalRecords { length = totalRecords - offset } off := offset // capture loop variable cwo := workflow.ChildWorkflowOptions{ WorkflowID: parentID + "/batch-" + fmt.Sprintf("%d", off), TaskQueue: TaskQueue, } cctx := workflow.WithChildOptions(ctx, cwo) futures = append(futures, workflow.ExecuteChildWorkflow(cctx, RecordBatchWorkflow, off, length)) } total := 0 for _, f := range futures { var n int if err := f.Get(ctx, &n); err != nil { return total, err } total += n } return total, nil } func RecordBatchWorkflow(ctx workflow.Context, offset int, length int) (int, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) processed := 0 for i := offset; i < offset+length; i++ { if err := workflow.ExecuteActivity(ctx, ProcessRecord, i).Get(ctx, nil); err != nil { return processed, err } processed++ } return processed, nil } ```

Fast/Slow Retries Go example

```go package downstream import ( "time" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) func FastSlowRetryWorkflow(ctx workflow.Context, request string) (string, error) { log := workflow.GetLogger(ctx) // Phase 1: fast retries fastCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{ InitialInterval: time.Second, BackoffCoefficient: 1.5, MaximumInterval: 30 * time.Second, MaximumAttempts: 10, }, }) var result string err := workflow.ExecuteActivity(fastCtx, CallDownstream, request).Get(fastCtx, &result) if err != nil { log.Warn("Fast retries exhausted — switching to slow retry phase", "request", request) // Phase 2: slow retries slowCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{ InitialInterval: 5 * time.Minute, BackoffCoefficient: 1.0, // MaximumAttempts defaults to 0 (unlimited) }, }) err = workflow.ExecuteActivity(slowCtx, CallDownstream, request).Get(slowCtx, &result) } return result, err } ``` This example shows a Workflow implementing fast retries with 1-second initial interval and 10 max attempts, transitioning to slow retries with 5-minute interval and unlimited attempts when fast phase is exhausted.

Go Fixed Count Retries example

package payments import ( "errors" "time" enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) func PaymentWorkflow(ctx workflow.Context, orderID string) (string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 3, }, } ctx = workflow.WithActivityOptions(ctx, ao) var result string err := workflow.ExecuteActivity(ctx, ChargePaymentAPI, orderID).Get(ctx, &result) if err != nil { var actErr *temporal.ActivityError if errors.As(err, &actErr) && actErr.RetryState() == enumspb.RETRY_STATE_MAXIMUM_ATTEMPTS_REACHED { workflow.GetLogger(ctx).Error("Payment failed: all 3 attempts exhausted", "orderID", orderID) } return "", err } return result, nil } This example shows how to set MaximumAttempts=3 on a RetryPolicy for a Go workflow and check the retry state when an error occurs.

Fixed Wall-Time Retries Go implementation

Example of enforcing a 2-minute SLA with per-attempt 30-second timeout: ```go func PaymentAuthWorkflow(ctx workflow.Context, transactionID string) (string, error) { ao := workflow.ActivityOptions{ ScheduleToCloseTimeout: 2 * time.Minute, // total budget StartToCloseTimeout: 30 * time.Second, // per attempt RetryPolicy: &temporal.RetryPolicy{ InitialInterval: 5 * time.Second, BackoffCoefficient: 1.5, MaximumInterval: 30 * time.Second, }, } ctx = workflow.WithActivityOptions(ctx, ao) var result string err := workflow.ExecuteActivity(ctx, AuthorizeTransaction, transactionID).Get(ctx, &result) if err != nil { var timeoutErr *temporal.TimeoutError if errors.As(err, &timeoutErr) && timeoutErr.TimeoutType() == enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE { workflow.GetLogger(ctx).Error( "Authorization failed — 2-minute SLA breached", "transactionID", transactionID, ) } return "", err } return result, nil } ```

Short SLA without per-attempt timeout Go

For a 30-second authorization window, omit StartToCloseTimeout and let ScheduleToCloseTimeout act as the only bound: ```go ao := workflow.ActivityOptions{ ScheduleToCloseTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{ InitialInterval: 3 * time.Second, BackoffCoefficient: 1.5, }, } ```

Local Activity Go API

In Go, use `workflow.WithLocalActivityOptions()` to set Local Activity options, then `workflow.ExecuteLocalActivity()` to execute the activity. Set `ScheduleToCloseTimeout` in the `LocalActivityOptions`.

Local Activity Go example

func TransactionWorkflow(ctx workflow.Context, req TransactionRequest) (Transaction, error) { localCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{ ScheduleToCloseTimeout: 10 * time.Second, }) // All three activities run in-process — no server round-trips. var tx Transaction if err := workflow.ExecuteLocalActivity(localCtx, ValidateTransaction, req).Get(localCtx, &tx); err != nil { return Transaction{}, err } if err := workflow.ExecuteLocalActivity(localCtx, ReserveFunds, tx).Get(localCtx, &tx); err != nil { return Transaction{}, err } if err := workflow.ExecuteLocalActivity(localCtx, SettleTransaction, tx).Get(localCtx, &tx); err != nil { return Transaction{}, err } return tx, nil }

Go Activity Heartbeat - basic progress tracking

Example in Go showing file processing with heartbeating every 100 lines. Uses activity.HasHeartbeatDetails(ctx) to check for prior progress, activity.GetHeartbeatDetails(ctx, &startLine) to retrieve the last line number, and activity.RecordHeartbeat(ctx, currentLine) to store progress.

Go Activity Heartbeat - cancellation handling

In Go, cancellation is detected by checking ctx.Done() after heartbeating. Use select { case <-ctx.Done(): cleanupResources(); return ctx.Err() default: } to handle cancellation. The Activity should call cleanupResources() before returning ctx.Err().

Go non-retryable error creation

In Go, use go.temporal.io/sdk/temporal.NewNonRetryableApplicationError() to create a non-retryable error at the throw site. Pass the error message and type name as parameters. Example: return "", temporal.NewNonRetryableApplicationError(fmt.Sprintf("order %s not found", orderID), "OrderNotFoundError", nil)

Go RetryPolicy NonRetryableErrorTypes field

In Go, set the NonRetryableErrorTypes field of a temporal.RetryPolicy to a []string slice of error type names that should never be retried. Assign this to ActivityOptions.RetryPolicy. Example: RetryPolicy: &temporal.RetryPolicy{NonRetryableErrorTypes: []string{"OrderNotFoundError", "ValidationError"}}}

Catching and handling ActivityError in Go workflows

In Go, call .Get() on the result of workflow.ExecuteActivity(). Use errors.As() to check if the error is a *temporal.ApplicationError and inspect its Type() method to route to the appropriate compensation or escalation path.

Frequent polling example in Go

```go // activities.go func DoPoll(ctx context.Context) (string, error) { for { activity.RecordHeartbeat(ctx) result, err := externalService.CheckStatus() if err != nil { return "", err } if result == "COMPLETED" { return result, nil } select { case <-ctx.Done(): return "", ctx.Err() case <-time.After(1 * time.Second): } } } // workflow.go func FrequentPollingWorkflow(ctx workflow.Context) (string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 60 * time.Second, HeartbeatTimeout: 2 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) var result string err := workflow.ExecuteActivity(ctx, DoPoll).Get(ctx, &result) return result, err } ``` This example shows a frequent polling Activity that loops indefinitely with heartbeats every iteration, and a Workflow that executes the Activity with a 60-second start-to-close timeout and 2-second heartbeat timeout.

Infrequent polling example in Go

```go // activities.go func DoPoll(ctx context.Context) (string, error) { result, err := externalService.CheckStatus() if err != nil { return "", err } if result != "COMPLETED" { return "", fmt.Errorf("service not ready, will retry") } return result, nil } // workflow.go func InfrequentPollingWorkflow(ctx workflow.Context) (string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 2 * time.Second, RetryPolicy: &temporal.RetryPolicy{ BackoffCoefficient: 1, InitialInterval: 60 * time.Second, }, } ctx = workflow.WithActivityOptions(ctx, ao) var result string err := workflow.ExecuteActivity(ctx, DoPoll).Get(ctx, &result) return result, err } ``` This example shows an infrequent polling Activity that performs a single poll and returns an error if the service is not ready, and a Workflow that executes the Activity with a fixed 60-second retry interval (backoff_coefficient=1).

Periodic sequence polling example in Go

```go // workflow.go func PollingChildWorkflow(ctx workflow.Context, pollingIntervalSeconds int) (string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) maxAttempts := 10 for i := 0; i < maxAttempts; i++ { var result string err := workflow.ExecuteActivity(ctx, DoPoll).Get(ctx, &result) if err != nil { return "", err } if result == "COMPLETED" { return result, nil } workflow.Sleep(ctx, time.Duration(pollingIntervalSeconds)*time.Second) } // Continue-as-new to prevent unbounded history return "", workflow.NewContinueAsNewError(ctx, PollingChildWorkflow, pollingIntervalSeconds) } func PeriodicPollingWorkflow(ctx workflow.Context) (string, error) { cwo := workflow.ChildWorkflowOptions{ WorkflowID: "ChildWorkflowPoll", } ctx = workflow.WithChildOptions(ctx, cwo) var result string err := workflow.ExecuteChildWorkflow(ctx, PollingChildWorkflow, 5).Get(ctx, &result) return result, err } ``` This example shows a Child Workflow that polls up to 10 times with a configurable interval between attempts, then calls Continue-As-New to start a fresh execution. The parent Workflow executes the Child Workflow with a specific workflow ID.

Priority Task Queues example in Go

```go we, err := c.ExecuteWorkflow( context.Background(), client.StartWorkflowOptions{ ID: "charge-customer-wf", TaskQueue: "my-task-queue", Priority: temporal.Priority{PriorityKey: 1}, }, ChargeCustomer, ) ``` This example shows how to set Workflow priority at start time using the Go SDK.

Set Activity priority in Go

```go ao := workflow.ActivityOptions{ StartToCloseTimeout: time.Minute, Priority: temporal.Priority{PriorityKey: 1}, } ctx = workflow.WithActivityOptions(ctx, ao) err := workflow.ExecuteActivity(ctx, ProcessPayment).Get(ctx, nil) ``` This example shows how to override an Activity's priority from the parent Workflow using the Go SDK.

Set Child Workflow priority in Go

```go cwo := workflow.ChildWorkflowOptions{ WorkflowID: "process-order-child", TaskQueue: "my-task-queue", Priority: temporal.Priority{PriorityKey: 2}, } ctx = workflow.WithChildOptions(ctx, cwo) err := workflow.ExecuteChildWorkflow(ctx, ProcessOrder).Get(ctx, nil) ``` This example shows how to set a Child Workflow's priority using the Go SDK.

Go task assignment workflow with Updates

```go // workflow.go type TaskWorkflow struct{} const MaxTasks = 10 func (w *TaskWorkflow) Run(ctx workflow.Context) error { tasks := []string{} err := workflow.SetUpdateHandlerWithOptions( ctx, "AssignTask", func(ctx workflow.Context, taskName string) (AssignmentResult, error) { assignmentID := uuid.New().String() tasks = append(tasks, taskName) return AssignmentResult{ AssignmentID: assignmentID, TaskName: taskName, TotalTasks: len(tasks), }, nil }, workflow.UpdateHandlerOptions{ Validator: func(taskName string) error { if len(tasks) >= MaxTasks { return fmt.Errorf("task limit reached") } return nil }, }, ) if err != nil { return err } err = workflow.SetQueryHandler(ctx, "GetTasks", func() ([]string, error) { return tasks, nil }) if err != nil { return err } workflow.GetSignalChannel(ctx, "").Receive(ctx, nil) return nil } ```

Resumable Activity Go implementation

Go Resumable Activity Workflow implementation: ```go package transfer import ( "fmt" "time" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) type TransferInput struct { FromAccount string ToAccount string Amount float64 } func TransferWorkflow(ctx workflow.Context, input TransferInput) (string, error) { status := "PENDING" if err := workflow.SetQueryHandler(ctx, "getStatus", func() (string, error) { return status, nil }); err != nil { return "", err } correctionCh := workflow.GetSignalChannel(ctx, "retryWithCorrection") approvalCh := workflow.GetSignalChannel(ctx, "approve") ao := workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, RetryPolicy: &temporal.RetryPolicy{MaximumAttempts: 3}, } actCtx := workflow.WithActivityOptions(ctx, ao) account := input.ToAccount correctionCount := 0 for { status = "TRANSFERRING" err := workflow.ExecuteActivity(actCtx, ExecuteTransfer, TransferInput{ FromAccount: input.FromAccount, ToAccount: account, Amount: input.Amount, }).Get(actCtx, nil) if err == nil { break } correctionCount++ if correctionCount > 5 { status = "FAILED" _ = workflow.UpsertSearchAttributes(ctx, map[string]interface{}{"TransferStatus": status}) return "", err } status = "AWAITING_CORRECTION" _ = workflow.UpsertSearchAttributes(ctx, map[string]interface{}{"TransferStatus": status}) workflow.GetLogger(ctx).Warn("Transfer failed — waiting for account correction", "to_account", account) var corrected string _ = workflow.Await(ctx, func() bool { return correctionCh.ReceiveAsync(&corrected) }) account = corrected } status = "AWAITING_APPROVAL" var approved bool _ = workflow.Await(ctx, func() bool { return approvalCh.ReceiveAsync(&approved) }) if approved { status = "COMPLETED" return fmt.Sprintf("Transfer of %.2f to %s completed", input.Amount, account), nil } status = "REJECTED" return "Transfer rejected by client", nil } ```

Signal with Start Go implementation

In Go, use client.SignalWithStartWorkflow() with parameters: context, workflow ID, signal name, signal object, StartWorkflowOptions struct (containing ID and TaskQueue), and the workflow function. The method atomically starts the workflow if needed and delivers the signal. Example: c.SignalWithStartWorkflow(ctx, "cart-"+cartID, "addItem", sig, opts, ShoppingCartWorkflow).

Saga pattern best practice: disconnected context for Go compensations

In Go, use NewDisconnectedContext to run compensation Activities after Workflow cancellation, since the original context is already cancelled.

Go Saga pattern implementation

func OpenAccountWorkflow(ctx workflow.Context, req OpenAccountRequest) error { var compensations []func() runCompensations := func() { for i := len(compensations) - 1; i >= 0; i-- { compensations[i]() } } if err := workflow.ExecuteActivity(ctx, CreateAccount, req).Get(ctx, nil); err != nil { return err } compensations = append(compensations, func() { _ = workflow.ExecuteActivity(ctx, ClearPostalAddresses, req).Get(ctx, nil) }) if err := workflow.ExecuteActivity(ctx, AddAddress, req).Get(ctx, nil); err != nil { runCompensations() return err } compensations = append(compensations, func() { _ = workflow.ExecuteActivity(ctx, RemoveClient, req).Get(ctx, nil) }) if err := workflow.ExecuteActivity(ctx, AddClient, req).Get(ctx, nil); err != nil { runCompensations() return err } compensations = append(compensations, func() { _ = workflow.ExecuteActivity(ctx, DisconnectBankAccounts, req).Get(ctx, nil) }) if err := workflow.ExecuteActivity(ctx, AddBankAccount, req).Get(ctx, nil); err != nil { runCompensations() return err } return nil } This example shows how to implement the Saga pattern in Go using a slice of closures and iterating from the end on error.

Sliding Window Go implementation example

Example Go implementation of Sliding Window pattern: ```go package main import ( "fmt" "strings" "time" enums "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/workflow" ) const CompletionSignal = "recordCompleted" func RecordProcessorWorkflow(ctx workflow.Context, recordID string) error { ao := workflow.ActivityOptions{StartToCloseTimeout: 30 * time.Second} ctx = workflow.WithActivityOptions(ctx, ao) if err := workflow.ExecuteActivity(ctx, ProcessRecord, recordID).Get(ctx, nil); err != nil { return err } parentID := workflow.GetInfo(ctx).ParentWorkflowExecution.ID err := workflow.SignalExternalWorkflow(ctx, parentID, "", CompletionSignal, recordID).Get(ctx, nil) if err != nil && strings.Contains(err.Error(), "not found") { return nil } return err } func SlidingWindowWorkflow(ctx workflow.Context, input SlidingWindowInput) (int, error) { windowSize := input.WindowSize if windowSize <= 0 { windowSize = WindowSize } recordIDs := input.RecordIDs parentID := workflow.GetInfo(ctx).WorkflowExecution.ID completedCh := workflow.GetSignalChannel(ctx, CompletionSignal) nextIndex := input.StartIndex totalProcessed := input.TotalProcessed dispatched := 0 active := input.Active startChild := func(recordID string) error { cwo := workflow.ChildWorkflowOptions{ WorkflowID: fmt.Sprintf("%s/record-%s", parentID, recordID), TaskQueue: TaskQueue, ParentClosePolicy: enums.PARENT_CLOSE_POLICY_ABANDON, } future := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, cwo), RecordProcessorWorkflow, recordID) return future.GetChildWorkflowExecution().Get(ctx, nil) } for nextIndex < len(recordIDs) { if active >= windowSize { completedCh.Receive(ctx, nil) totalProcessed++ active-- } if err := startChild(recordIDs[nextIndex]); err != nil { return 0, err } nextIndex++ dispatched++ active++ if dispatched >= windowSize { return 0, workflow.NewContinueAsNewError(ctx, SlidingWindowWorkflow, SlidingWindowInput{ RecordIDs: recordIDs, WindowSize: windowSize, StartIndex: nextIndex, TotalProcessed: totalProcessed, Active: active, }) } } for active > 0 { completedCh.Receive(ctx, nil) totalProcessed++ active-- } return totalProcessed, nil } ```

Updatable Timer Go implementation

```go // updatable_timer.go func sleepUntil(ctx workflow.Context, wakeUpTime time.Time, wakeUpChannel workflow.ReceiveChannel) error { for { timerCtx, cancelTimer := workflow.WithCancel(ctx) duration := wakeUpTime.Sub(workflow.Now(ctx)) if duration <= 0 { cancelTimer() break } timer := workflow.NewTimer(timerCtx, duration) selector := workflow.NewSelector(ctx) timerFired := false selector.AddFuture(timer, func(f workflow.Future) { timerFired = true }) selector.AddReceive(wakeUpChannel, func(c workflow.ReceiveChannel, more bool) { c.Receive(ctx, &wakeUpTime) // Cancel the current timer so it can be recreated with the new deadline cancelTimer() }) selector.Select(ctx) if timerFired { break // Timer expired } // Signal received with new wakeUpTime, loop to recalculate } return nil } ```

Updatable Timer pitfall: not cancelling timers in Go

In the Go SDK, always cancel the previous timer (via workflow.WithCancel) before creating a new one. Uncancelled timers wake up the Workflow unnecessarily, creating extra Worker load.

Go Worker-Specific Task Queues example workflow

```go // workflow.go func FileProcessingWorkflow(ctx workflow.Context, source string, destination string) error { defaultOptions := workflow.ActivityOptions{ StartToCloseTimeout: 20 * time.Second, } defaultCtx := workflow.WithActivityOptions(ctx, defaultOptions) var activities *StoreActivities var downloaded TaskQueueFileNamePair err := workflow.ExecuteActivity(defaultCtx, activities.Download, source).Get(ctx, &downloaded) if err != nil { return err } hostOptions := workflow.ActivityOptions{ TaskQueue: downloaded.HostTaskQueue, ScheduleToStartTimeout: 10 * time.Second, StartToCloseTimeout: 20 * time.Second, } hostCtx := workflow.WithActivityOptions(ctx, hostOptions) var processed string err = workflow.ExecuteActivity(hostCtx, activities.Process, downloaded.FileName).Get(ctx, &processed) if err != nil { return err } return workflow.ExecuteActivity(hostCtx, activities.Upload, processed, destination).Get(ctx, nil) } ``` This example shows a workflow that downloads a file on any worker, then processes and uploads it on the same host-specific worker.

Go Worker-Specific Task Queues example activity

```go // activities.go type TaskQueueFileNamePair struct { HostTaskQueue string FileName string } type StoreActivities struct { HostSpecificTaskQueue string } func (a *StoreActivities) Download(ctx context.Context, source string) (*TaskQueueFileNamePair, error) { localFile, err := downloadToLocalDisk(source) if err != nil { return nil, err } return &TaskQueueFileNamePair{ HostTaskQueue: a.HostSpecificTaskQueue, FileName: localFile, }, nil } func (a *StoreActivities) Process(ctx context.Context, fileName string) (string, error) { return processLocalFile(fileName) } func (a *StoreActivities) Upload(ctx context.Context, fileName string, destination string) error { return uploadFromLocalDisk(fileName, destination) } ``` The download method returns the host-specific Task Queue name alongside the file path. The process and upload methods operate on local files, which are guaranteed to exist because they run on the same host.

Go Worker-Specific Task Queues example worker setup

```go // worker/main.go func main() { c, err := client.Dial(client.Options{}) if err != nil { log.Fatalln("Unable to create client", err) } defer c.Close() defaultTaskQueue := "FileProcessing" hostTaskQueue := fmt.Sprintf("FileProcessing-%s-%s", getHostName(), uuid.New().String()) activities := &StoreActivities{HostSpecificTaskQueue: hostTaskQueue} defaultWorker := worker.New(c, defaultTaskQueue, worker.Options{}) defaultWorker.RegisterWorkflow(FileProcessingWorkflow) defaultWorker.RegisterActivity(activities) hostWorker := worker.New(c, hostTaskQueue, worker.Options{}) hostWorker.RegisterActivity(activities) err = defaultWorker.Start() if err != nil { log.Fatalln("Unable to start default worker", err) } err = hostWorker.Start() if err != nil { log.Fatalln("Unable to start host worker", err) } // Block until interrupted select {} } ``` Each Worker registers with both the default Task Queue and its own host-specific Task Queue. The host-specific queue name includes hostname and UUID for uniqueness.

Go SDK FailureConverter configuration

With the Temporal Go SDK, you can configure a FailureConverter by adding a FailureConverter parameter to client.Options{} when calling client.Dial(). Use temporal.NewDefaultFailureConverter(temporal.DefaultFailureConverterOptions{EncodeCommonAttributes: true}) to enable encoding of failure attributes.

Default Data Converter Payload Converter order in Go

The default Data Converter logic in Temporal is implemented as a Composite Data Converter with the following order of Payload Converters: NewNilPayloadConverter(), NewByteSlicePayloadConverter(), NewProtoJSONPayloadConverter(), NewProtoPayloadConverter(), NewJSONPayloadConverter().

Task Queue constant definition - Go example

In Go, define a Task Queue name constant in a package: const TaskQueueName = "my-taskqueue-name". Reference this constant as app.TaskQueueName in both the workflow client code (in StartWorkflowOptions.TaskQueue) and worker configuration (in worker.New()).

Workflow Definition syntax in Rust

A Workflow Definition in Rust uses the #[workflow] macro on a struct and #[workflow_methods] macro on the impl block. The struct has an #[init] constructor and an #[run] async method returning WorkflowResult. Example uses temporalio_macros crate.

Workflow Definition syntax in Go

A Workflow Definition in Go is typically implemented as a function with signature: func YourBasicWorkflow(ctx workflow.Context) error { ... }

Workflow Definition syntax in .NET

A Workflow Definition in C# and .NET uses a class with [Workflow] attribute containing an async method with [WorkflowRun] attribute. Example: [Workflow] public class YourBasicWorkflow { [WorkflowRun] public async Task<string> workflowExample(string param) { ... } }

Eager Workflow Start Go example

func main() { c, err := client.Dial(client.Options{}) if err != nil { log.Fatalln("Unable to create Temporal client:", err) } defer c.Close() w := worker.New(c, TaskQueue, worker.Options{}) w.RegisterWorkflow(TransactionWorkflow) w.RegisterActivity(ValidateTransaction) w.RegisterActivity(SettleTransaction) if err := w.Start(); err != nil { log.Fatalln("Unable to start worker:", err) } defer w.Stop() run, err := c.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ ID: "eager-workflow-start-demo", TaskQueue: TaskQueue, EnableEagerStart: true, }, TransactionWorkflow, TransactionRequest{Amount: 100.00, Currency: "USD"}) if err != nil { log.Fatalln("Failed to start workflow:", err) } var result Transaction if err := run.Get(context.Background(), &result); err != nil { log.Fatalln("Workflow failed:", err) } fmt.Printf("Transaction complete: ID=%s Status=%s\n", result.ID, result.Status) } This example shows starting a non-blocking Worker with w.Start() in the same process, then executing a Workflow with EnableEagerStart: true to dispatch the first WorkflowTask inline.

Early Return + Local Activities Go example

```go // workflows.go func TransactionWorkflow(ctx workflow.Context, req TransactionRequest) error { var tx Transaction var initDone bool var initErr error // Register Update handler — returns to the client as soon as Phase 1 is done. if err := workflow.SetUpdateHandler(ctx, "getResult", func(ctx workflow.Context, r TransactionRequest) (Transaction, error) { _ = workflow.Await(ctx, func() bool { return initDone }) return tx, initErr }); err != nil { return err } // Phase 1: Local Activities — in-process, zero server round-trips. localCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{ ScheduleToCloseTimeout: 5 * time.Second, }) if err := workflow.ExecuteLocalActivity(localCtx, ValidateTransaction, req).Get(localCtx, &tx); err == nil { initErr = workflow.ExecuteLocalActivity(localCtx, InitTransaction, tx).Get(localCtx, &tx) } else { initErr = err } initDone = true activityCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, }) if initErr != nil { // Phase 2 (cancel): regular Activity runs in the background. return workflow.ExecuteActivity(activityCtx, CancelTransaction, tx).Get(activityCtx, nil) } // Phase 2 (complete): regular Activity runs in the background. return workflow.ExecuteActivity(activityCtx, CompleteTransaction, tx).Get(activityCtx, nil) } ```

Go Pattern 1 Signal-with-Start starter example

```go func main() { c, err := client.Dial(client.Options{HostPort: "localhost:7233"}) if err != nil { log.Fatalln("Unable to create client:", err) } defer c.Close() ctx := context.Background() orderID := fmt.Sprintf("order-%d", time.Now().UnixMilli()) workflowID := "order-" + orderID order := OrderInput{OrderID: orderID, Amount: 99.99} payment := PaymentPayload{ PaymentID: fmt.Sprintf("pay-%d", time.Now().UnixMilli()), Amount: 99.99, } fmt.Printf("Sending webhook for order %s\n", orderID) // Signal-with-Start: atomically starts the workflow (if not running) and // delivers the payment signal — this is exactly what your HTTP handler would do. we, err := c.SignalWithStartWorkflow( ctx, workflowID, SignalName, payment, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: TaskQueue, }, OrderWorkflow, order, ) if err != nil { log.Fatalln("SignalWithStart failed:", err) } fmt.Printf("Webhook signal sent: %s\n", payment.PaymentID) var result string if err := we.Get(ctx, &result); err != nil { log.Fatalln("Workflow result failed:", err) } fmt.Printf("Order completed: %s\n", result) } ``` This Go example demonstrates SignalWithStartWorkflow which atomically creates the workflow if needed and delivers the signal.

Go Pattern 2 delayed outbound callback example

```go func DelayedCallbackWorkflow(ctx workflow.Context, input CallbackInput) error { workflow.GetLogger(ctx).Info("Sleeping before callback", "delay", input.DelaySeconds, "url", input.CallbackURL) // Durable sleep — survives worker restarts, server restarts, everything if err := workflow.Sleep(ctx, time.Duration(input.DelaySeconds)*time.Second); err != nil { return err } // Fire the outbound callback; Temporal retries on HTTP failure ao := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 5 * time.Minute, }) return workflow.ExecuteActivity(ao, SendWebhookCallback, input).Get(ao, nil) } ``` This Go example uses workflow.Sleep() for a durable delay followed by an activity to send the callback.

Go Pattern 1 inbound webhook example

```go func OrderWorkflow(ctx workflow.Context, order OrderInput) (string, error) { workflow.GetLogger(ctx).Info("Order waiting for payment webhook", "order_id", order.OrderID) var payment *PaymentPayload // Block until the inbound webhook signal arrives (or timeout after 24 hours) selector := workflow.NewSelector(ctx) timerFired := false timerCtx, cancelTimer := workflow.WithCancel(ctx) timer := workflow.NewTimer(timerCtx, 24*time.Hour) signalCh := workflow.GetSignalChannel(ctx, SignalName) selector.AddReceive(signalCh, func(ch workflow.ReceiveChannel, more bool) { ch.Receive(ctx, &payment) cancelTimer() }) selector.AddFuture(timer, func(f workflow.Future) { if err := f.Get(ctx, nil); err == nil { timerFired = true } }) selector.Select(ctx) if timerFired || payment == nil { return "Order " + order.OrderID + ": timed out waiting for payment", nil } ao := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, }) var result string err := workflow.ExecuteActivity(ao, ProcessPayment, payment).Get(ao, &result) return result, err } ``` This Go example uses workflow.NewSelector to wait for either a signal or timeout, with a 24-hour maximum wait.

Go Pattern 3 async activity completion example

```go // SubmitJob submits a job and returns immediately; the activity completes // when the external callback arrives and calls CompleteAsyncActivity. func SubmitJob(ctx context.Context, input JobInput) (string, error) { info := activity.GetInfo(ctx) // Persist the task token so the callback handler can retrieve it by job ID taskToken := info.TaskToken jobID := fmt.Sprintf("job-%d", info.StartedTime.UnixMilli()) if err := persistTaskToken(jobID, hex.EncodeToString(taskToken)); err != nil { return "", err } fmt.Printf("Job %s submitted; waiting for async callback\n", jobID) // Return ErrResultPending to tell Temporal not to mark the activity complete yet return "", activity.ErrResultPending } // CompleteJob is called by your webhook callback handler to unblock the workflow. func CompleteJob(ctx context.Context, c client.Client, jobID string, result string) error { tokenHex, err := loadTaskToken(jobID) if err != nil { return err } taskToken, _ := hex.DecodeString(tokenHex) return c.CompleteActivity(ctx, taskToken, result, nil) } ``` This Go example shows Pattern 3 where SubmitJob returns activity.ErrResultPending to signal async completion, and CompleteJob uses the task token to complete the activity.

Set fairness key and weight at Workflow start in Go

Set FairnessKey and FairnessWeight in the Priority field of client.StartWorkflowOptions. Example: ```go we, err := c.ExecuteWorkflow( context.Background(), client.StartWorkflowOptions{ ID: "process-order-wf", TaskQueue: "my-task-queue", Priority: temporal.Priority{ FairnessKey: "tenant-a", FairnessWeight: 2.0, }, }, ProcessOrder, ) ```

Set fairness key and weight on Activities in Go

Set FairnessKey and FairnessWeight in the Priority field of workflow.ActivityOptions. Example: ```go ao := workflow.ActivityOptions{ StartToCloseTimeout: time.Minute, Priority: temporal.Priority{ FairnessKey: "tenant-a", FairnessWeight: 2.0, }, } ctx = workflow.WithActivityOptions(ctx, ao) err := workflow.ExecuteActivity(ctx, ProcessForTenant, req).Get(ctx, nil) ```

Use priority and fairness together in Go

Set both PriorityKey and FairnessKey in the Priority field of StartWorkflowOptions. Example: ```go we, err := c.ExecuteWorkflow( context.Background(), client.StartWorkflowOptions{ ID: "charge-customer-wf", TaskQueue: "my-task-queue", Priority: temporal.Priority{ PriorityKey: 1, FairnessKey: "tenant-a", FairnessWeight: 2.0, }, }, ChargeCustomer, ) ```

Event Accumulator Go implementation

Go implementation uses workflow.GetSignalChannel to obtain named signal channels, then builds a Selector per iteration with a NewTimer future and two channel receivers. Cancels the old timer whenever a signal arrives before it fires.

Event Accumulator Go example workflow

func AccumulatorWorkflow(ctx workflow.Context, bucketKey string, items []OrderItem, seenKeys []string) (string, error) { seenSet := make(map[string]bool) for _, k := range seenKeys { seenSet[k] = true } accumulated := append([]OrderItem{}, items...) addItemCh := workflow.GetSignalChannel(ctx, "add-item") flushCh := workflow.GetSignalChannel(ctx, "flush") flushRequested := false for { // Drain any signals buffered before this iteration for { var item OrderItem if !addItemCh.ReceiveAsync(&item) { break } if item.OrderID == bucketKey && !seenSet[item.ItemID] { seenSet[item.ItemID] = true accumulated = append(accumulated, item) } } var voidFlush interface{} if flushCh.ReceiveAsync(&voidFlush) { flushRequested = true } if flushRequested { break } if workflow.GetInfo(ctx).GetContinueAsNewSuggested() { keys := make([]string, 0, len(seenSet)) for k := range seenSet { keys = append(keys, k) } sort.Strings(keys) // deterministic order for replay return "", workflow.NewContinueAsNewError(ctx, AccumulatorWorkflow, bucketKey, accumulated, keys) } // Sliding window: wait for a signal or let the inactivity timer fire timedOut := false timerCtx, cancelTimer := workflow.WithCancel(ctx) timer := workflow.NewTimer(timerCtx, maxAwaitTime) selector := workflow.NewSelector(ctx) selector.AddFuture(timer, func(f workflow.Future) { timedOut = true }) selector.AddReceive(addItemCh, func(c workflow.ReceiveChannel, _ bool) { var item OrderItem c.Receive(ctx, &item) if item.OrderID == bucketKey && !seenSet[item.ItemID] { seenSet[item.ItemID] = true accumulated = append(accumulated, item) } }) selector.AddReceive(flushCh, func(c workflow.ReceiveChannel, _ bool) { var void interface{} c.Receive(ctx, &void) flushRequested = true }) selector.Select(ctx) cancelTimer() // no-op if timer already fired; cancels timer if a signal arrived if timedOut || flushRequested { break } } ao := workflow.ActivityOptions{StartToCloseTimeout: 10 * time.Second} actCtx := workflow.WithActivityOptions(ctx, ao) var result string if err := workflow.ExecuteActivity(actCtx, ProcessItems, bucketKey, accumulated).Get(ctx, &result); err != nil { return "", err } workflow.GetLogger(ctx).Info("Processed order batch", "bucketKey", bucketKey, "count", len(accumulated)) return result, nil }

Give your agent this brain