Skip to main content

Cross Namespace (XNS)

Cross-Namespace (XNS) functionality enables workflows in one Temporal namespace or cluster to invoke workflows, send signals, execute queries, and perform updates on workflows running in different namespaces or clusters. The protoc-gen-go-temporal plugin generates XNS helper functions that wrap these operations as activities, providing type-safe, cross-namespace communication.

XNS Configuration​

Enable XNS functionality by adding xns configuration blocks to your protobuf temporal method options.

example.proto
syntax="proto3";

package example.v1;

import "temporal/v1/temporal.proto";

service Example {
  option (temporal.v1.service) = {
    task_queue: "example-v1"
  };

  rpc CreateFoo(CreateFooRequest) returns (CreateFooResponse) {
    option (temporal.v1.workflow) = {
      execution_timeout: { seconds: 3600 }
      id: 'create-foo/${! name.slug() }'
      xns: {
        heartbeat_interval: { seconds: 10 }
        heartbeat_timeout: { seconds: 20 }
        start_to_close_timeout: { seconds: 3630 }
      }
    };
  }

  rpc GetFooProgress(google.protobuf.Empty) returns (GetFooProgressResponse) {
    option (temporal.v1.query) = {
      xns: {
        heartbeat_interval: { seconds: 5 }
        heartbeat_timeout: { seconds: 15 }
        start_to_close_timeout: { seconds: 60 }
      }
    };
  }

  rpc SetFooProgress(SetFooProgressRequest) returns (google.protobuf.Empty) {
    option (temporal.v1.signal) = {
      xns: {
        heartbeat_interval: { seconds: 5 }
        heartbeat_timeout: { seconds: 15 }
        start_to_close_timeout: { seconds: 60 }
      }
    };
  }

  rpc UpdateFooProgress(SetFooProgressRequest) returns (GetFooProgressResponse) {
    option (temporal.v1.update) = {
      id: 'update-progress/${! progress.string() }'
      xns: {
        heartbeat_interval: { seconds: 5 }
        heartbeat_timeout: { seconds: 15 }
        start_to_close_timeout: { seconds: 60 }
      }
    };
  }
}

Worker Setup​

Configure workers in both the calling namespace (source) and target namespace to enable XNS communication.

target_worker.go
package main

import (
    "log"

    examplev1 "path/to/gen/example/v1"
    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
)

func main() {
    // Connect to target namespace
    targetClient, err := client.Dial(client.Options{
        HostPort:  "localhost:7233",
        Namespace: "target-namespace",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer targetClient.Close()

    // Create worker for target namespace
    targetWorker := worker.New(targetClient, examplev1.ExampleTaskQueue, worker.Options{})

    // Register actual workflow and activity implementations
    examplev1.RegisterExampleWorkflows(targetWorker, &ExampleWorkflows{})
    examplev1.RegisterExampleActivities(targetWorker, &ExampleActivities{})

    log.Println("Target namespace worker starting...")
    err = targetWorker.Run(worker.InterruptCh())
    if err != nil {
        log.Fatal(err)
    }
}

XNS Workflow Execution​

Execute workflows across namespaces using the generated XNS helpers.

cross_namespace_workflow.go
package main

import (
    "fmt"

    examplev1 "path/to/gen/example/v1"
    examplev1xns "path/to/gen/example/v1/examplev1xns"
    "go.temporal.io/sdk/workflow"
)

type CallingWorkflow struct {
    *CallingWorkflowInput
}

func (w *CallingWorkflow) Execute(ctx workflow.Context) (*CallingWorkflowResponse, error) {
    // Execute workflow in target namespace synchronously
    result, err := examplev1xns.CreateFoo(ctx, &examplev1.CreateFooRequest{
        Name: "cross-namespace-foo",
    })
    if err != nil {
        return nil, fmt.Errorf("cross-namespace workflow failed: %w", err)
    }

    workflow.GetLogger(ctx).Info("Cross-namespace workflow completed", 
        "result", result.Foo.Name)

    return &CallingWorkflowResponse{
        RemoteResult: result,
    }, nil
}

XNS Communication​

Send signals, execute queries, and perform updates on workflows running in different namespaces.

xns_signals.go
func (w *CallingWorkflow) Execute(ctx workflow.Context) error {
    // Start remote workflow
    run, err := examplev1xns.CreateFooAsync(ctx, &examplev1.CreateFooRequest{
        Name: "signal-target",
    })
    if err != nil {
        return err
    }

    // Send signals to remote workflow
    for progress := 0.2; progress <= 1.0; progress += 0.2 {
        workflow.Sleep(ctx, time.Second*5)

        err = run.SetFooProgress(ctx, &examplev1.SetFooProgressRequest{
            Progress: float32(progress),
        })
        if err != nil {
            workflow.GetLogger(ctx).Error("Failed to send signal", "error", err)
        } else {
            workflow.GetLogger(ctx).Info("Sent progress signal", "progress", progress)
        }
    }

    // Wait for completion
    result, err := run.Get(ctx)
    if err != nil {
        return err
    }

    workflow.GetLogger(ctx).Info("Remote workflow completed", "result", result)
    return nil
}

XNS Options and Configuration​

Configure XNS operations using the generated options builders.

xns_workflow_options.go
func (w *CallingWorkflow) Execute(ctx workflow.Context) error {
    // Configure XNS workflow with custom options
    run, err := examplev1xns.CreateFooAsync(ctx, &examplev1.CreateFooRequest{
        Name: "configured-foo",
    }, examplev1xns.NewCreateFooWorkflowOptions().
        WithDetached(false).  // Wait for completion (default)
        WithHeartbeatInterval(time.Second * 15).
        WithParentClosePolicy(enums.PARENT_CLOSE_POLICY_REQUEST_CANCEL).
        WithActivityOptions(workflow.ActivityOptions{
            StartToCloseTimeout: time.Minute * 10,
            RetryPolicy: &temporal.RetryPolicy{
                MaximumAttempts: 3,
            },
        }),
    )
    if err != nil {
        return err
    }

    result, err := run.Get(ctx)
    if err != nil {
        return err
    }

    workflow.GetLogger(ctx).Info("Configured XNS workflow completed", "result", result)
    return nil
}

XNS Error Handling​

Handle errors from cross-namespace operations with built-in error conversion and custom error handling.

xns_error_handling.go
import (
    "errors"
    
    "github.com/cludden/protoc-gen-go-temporal/pkg/xns"
    examplev1xns "path/to/gen/example/v1/examplev1xns"
    "go.temporal.io/sdk/temporal"
)

func (w *CallingWorkflow) Execute(ctx workflow.Context) error {
    run, err := examplev1xns.CreateFooAsync(ctx, &examplev1.CreateFooRequest{
        Name: "error-prone-foo",
    })
    if err != nil {
        return fmt.Errorf("failed to start XNS workflow: %w", err)
    }

    result, err := run.Get(ctx)
    if err != nil {
        // Handle different types of XNS errors
        var applicationError *temporal.ApplicationError

        if errors.As(err, &applicationError) {
            // Extract XNS error details
            errorCode := xns.Code(err)
            isNonRetryable := xns.IsNonRetryable(err)
            underlyingErr := xns.Unwrap(err)

            workflow.GetLogger(ctx).Error("XNS workflow failed",
                "error_code", errorCode,
                "non_retryable", isNonRetryable,
                "underlying_error", underlyingErr)

            // Handle specific error types
            switch errorCode {
            case "WORKFLOW_ALREADY_EXISTS":
                // Handle workflow already exists
                return fmt.Errorf("remote workflow already exists: %w", err)
            case "WORKFLOW_NOT_FOUND":
                // Handle workflow not found
                return fmt.Errorf("remote workflow not found: %w", err)
            default:
                return fmt.Errorf("remote workflow error: %w", err)
            }
        }

        return fmt.Errorf("XNS workflow execution failed: %w", err)
    }

    workflow.GetLogger(ctx).Info("XNS workflow completed successfully", "result", result)
    return nil
}

Advanced XNS Patterns​

Signal/Update with Start​

Start remote workflows with initial signals or updates for atomic initialization.

xns_signal_with_start.go
func (w *CallingWorkflow) Execute(ctx workflow.Context) error {
    // Start remote workflow with initial signal
    run, err := examplev1xns.CreateFooWithSetFooProgressAsync(ctx,
        &examplev1.CreateFooRequest{Name: "signal-start-foo"},
        &examplev1.SetFooProgressRequest{Progress: 0.25},
    )
    if err != nil {
        return fmt.Errorf("failed to start workflow with signal: %w", err)
    }

    // Remote workflow starts with initial progress of 0.25
    result, err := run.Get(ctx)
    if err != nil {
        return err
    }

    workflow.GetLogger(ctx).Info("Workflow started with signal completed", "result", result)
    return nil
}

Testing XNS Workflows​

Test cross-namespace workflows using Temporal's testing framework with XNS mocking.

xns_workflow_test.go
package main

import (
    "testing"

    "github.com/stretchr/testify/mock"
    "github.com/stretchr/testify/require"
    "go.temporal.io/sdk/testsuite"

    examplev1 "path/to/gen/example/v1"
    examplev1xns "path/to/gen/example/v1/examplev1xns"
)

func TestXNSWorkflow(t *testing.T) {
    suite := &testsuite.WorkflowTestSuite{}
    env := suite.NewTestWorkflowEnvironment()

    // Register local workflows
    RegisterCallingWorkflows(env, &CallingWorkflows{})

    // Mock XNS activities
    env.OnActivity(examplev1xns.CreateFooActivityName, mock.Anything, mock.Anything).
        Return(&examplev1.CreateFooResponse{
            Foo: &examplev1.Foo{
                Name:   "test-xns-foo",
                Status: examplev1.Foo_FOO_STATUS_READY,
            },
        }, nil)

    // Execute workflow that uses XNS
    env.ExecuteWorkflow("CallingWorkflow", &CallingWorkflowRequest{
        Name: "test-caller",
    })

    require.True(t, env.IsWorkflowCompleted())
    require.NoError(t, env.GetWorkflowError())

    var result CallingWorkflowResponse
    err := env.GetWorkflowResult(&result)
    require.NoError(t, err)
    require.Equal(t, "test-xns-foo", result.RemoteResult.Foo.Name)
}