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)
    }
}

Dynamic Client Selection​

By default, Register<Service>Activities targets a single, statically-configured client (and therefore a single cluster/namespace). To route each invocation to a different cluster or namespace based on the calling context or request payload, supply a client provider instead of (or in addition to) a static client.

A client provider is a generated per-service function type:

type ExampleClientProvider func(ctx context.Context, in *ExampleClientProviderInput) (examplev1.ExampleClient, error)

The ExampleClientProviderInput describes the invocation so the provider can select a client:

type ExampleClientProviderInput struct {
    // ActivityName is the fully-qualified name of the xns activity, e.g. "example.v1.Example.CreateFoo"
    ActivityName string
    // WorkflowID / RunID identify the target workflow execution, when known
    WorkflowID string
    RunID      string
    // Requests holds the unmarshalled request message(s): zero for cancel/get, one for most
    // operations, and two for the signal-with-start and update-with-start variants
    Requests []proto.Message
}

// Request returns the primary request message, or nil when there is none
func (in *ExampleClientProviderInput) Request() proto.Message

Provide it either via a dedicated register function or via the options builder:

// Route each invocation to a per-tenant namespace based on the request payload
clients := map[string]examplev1.ExampleClient{ /* ... */ }

examplev1xns.RegisterExampleActivitiesWithClientProvider(callingWorker,
    func(ctx context.Context, in *examplev1xns.ExampleClientProviderInput) (examplev1.ExampleClient, error) {
        if req, ok := in.Request().(*examplev1.CreateFooRequest); ok {
            if c, ok := clients[req.GetTenant()]; ok {
                return c, nil
            }
        }
        return nil, fmt.Errorf("no client for activity %s", in.ActivityName)
    },
)

An error returned from the provider is converted using the same error converter applied to all xns activity errors, and the activity fails (subject to its retry policy). The existing static form — RegisterExampleActivities(worker, examplev1.NewExampleClient(targetClient)) — continues to work unchanged and is equivalent to a provider that always returns the given client.

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
}

The client.StartWorkflowOptions value passed to WithStartWorkflow is serialized as protobuf, sent to the remote namespace as activity input, and rebuilt on the other side, so every supported field survives the boundary: workflow id and id reuse/conflict policies, task queue, execution/run/task timeouts, retry policy, memo, search attributes (both untyped and typed), eager start, start delay, cron schedule, static summary and details, priority, and versioning override. Values that arrive this way take precedence over the schema defaults declared on the remote workflow.

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)
}