design-patterns/go

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

MapReduce Tree Go implementation

Go implementation with LeafWorkflow and NodeWorkflow functions. LeafWorkflow executes ProcessLeaf Activity with StartToCloseTimeout and signals result using workflow.SignalExternalWorkflow(). NodeWorkflow gets signal channel with workflow.GetSignalChannel(ctx, ResultSignal), checks len(records) against LeafThreshold, starts child workflows with workflow.ExecuteChildWorkflow(), receives signals in loop, and signals aggregated results to parent.

Pick First pattern implementation in Go

In Go, the Pick First pattern uses `workflow.NewSelector()` with `AddFuture()` callbacks to race Activities. A shared cancellable context is created with `workflow.WithCancel(ctx)`. Activities are started with `workflow.ExecuteActivity()` using the child context. The selector waits for the first Activity with `selector.Select(ctx)`. After the first result is captured, remaining Activities are cancelled by calling `cancelHandler()`. To wait for cancellation cleanup, set `WaitForCancellation: true` in ActivityOptions and call `Get` on all futures after cancelling.

Retry Alerting via Metrics implementation in Go

```go package downstream import ( "context" "go.temporal.io/sdk/activity" ) const alertThreshold = 5 func CallDownstreamService(ctx context.Context, endpoint string) (string, error) { info := activity.GetInfo(ctx) if info.Attempt > alertThreshold { activity.GetMetricsHandler(ctx). Counter("high_activity_error_count"). Inc(1) } response, err := downstream.Call(endpoint) if err != nil { return "", err } return response.Data, nil } ``` This example shows reading the attempt number using activity.GetInfo and emitting a counter metric via activity.GetMetricsHandler.

Retry Alerting with dimension tags in Go

```go if info.Attempt > alertThreshold { activity.GetMetricsHandler(ctx). WithTags(map[string]string{ "activity_type": info.ActivityType.Name, "endpoint": endpoint, }). Counter("high_activity_error_count"). Inc(1) } ``` Add tags to the metric using WithTags to identify which Activity type and endpoint are producing high attempt counts.

Go SDK error and panic behavior in message handlers

In Go, returning an error from a message handler behaves like an Application Failure in other SDKs. Panics behave like non-Application Failure exceptions in other languages, causing a Workflow Task Failure.

Go SDK StartWorkflowOptions default behavior

The default StartWorkflowOptions behavior in the Go SDK is to not return an error when a new Workflow Execution is attempted with the same Workflow ID as an Open Workflow Execution. Instead, it returns a WorkflowRun instance representing the current or last run of the Open Workflow Execution. To return the 'Workflow execution already started' error, set WorkflowExecutionErrorWhenAlreadyStarted to true.

Parallel Execution in Go

In Go, workflow.ExecuteActivity() returns Future objects. Call .Get() on each Future to collect results.

Go parallel Activities example

func ProcessInParallel(ctx workflow.Context, items []string) ([]string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) futures := make([]workflow.Future, len(items)) for i, item := range items { futures[i] = workflow.ExecuteActivity(ctx, Process, item) } results := make([]string, len(items)) for i, future := range futures { if err := future.Get(ctx, &results[i]); err != nil { return nil, err } } return results, nil } This example starts one Activity per item in a list and waits for all of them to complete.

Go batch processing example

func ProcessBatch(ctx workflow.Context, items []string, maxParallel int) ([]string, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) var results []string for i := 0; i < len(items); i += maxParallel { end := i + maxParallel if end > len(items) { end = len(items) } batch := items[i:end] futures := make([]workflow.Future, len(batch)) for j, item := range batch { futures[j] = workflow.ExecuteActivity(ctx, Process, item) } for _, future := range futures { var result string if err := future.Get(ctx, &result); err != nil { return nil, err } results = append(results, result) } } return results, nil } This example implements controlled parallelism by processing items in batches.

Error handling in parallel execution - Go

func ProcessWithErrorHandling(ctx workflow.Context, items []string) ([]Result, error) { ao := workflow.ActivityOptions{ StartToCloseTimeout: 30 * time.Second, } ctx = workflow.WithActivityOptions(ctx, ao) futures := make([]workflow.Future, len(items)) for i, item := range items { futures[i] = workflow.ExecuteActivity(ctx, Process, item) } results := make([]Result, len(items)) for i, future := range futures { var output string if err := future.Get(ctx, &output); err != nil { results[i] = Result{Item: items[i], Error: err.Error()} } else { results[i] = Result{Item: items[i], Output: output} } } return results, nil } This example checks each Future.Get() call individually for errors so that individual failures do not prevent other Activities from completing.

Give your agent this brain