Skip to main content

Test Temporal Workers in Sandboxes

This tutorial deploys a small money-transfer app (web UI, Temporal server, Python, TypeScript, Go, or Java worker) to your cluster and shows how a Sandbox tests a changed worker version against the language's shared Temporal task queue, with no duplicated servers, no per-environment queues, and no crosstalk between tests.

The example is implemented in Python, TypeScript, Go, and Java. Choose the worker language that matches your stack; the Python tab is selected by default.

git clone https://github.com/signadot/examples.git
cd examples/temporal-tutorial

Full source: signadot/examples/temporal-tutorial

How Temporal changes the isolation problem​

Request routing works for synchronous traffic because something in the path (sidecar, proxy) chooses a destination per request. Temporal is different: a workflow is submitted once to a task queue; workers poll that queue independently and pull tasks at their own pace. Deploy a changed worker version next to the baseline and both receive tasks, causing either split execution (share a queue, arbitrary partition of tasks) or duplication (separate queues, both versions process everything).

Selective task routing extends sandbox isolation across the async hop using interceptors:

  1. The client stamps a routing key onto the workflow submission via OpenTelemetry baggage in the task headers (_tracer-data).
  2. Each worker (baseline and sandboxed) polls the same task queue.
  3. A workflow interceptor reads the routing key from the task headers, consults the Signadot Routes API to learn which sandboxes are live, and rejects tasks that do not belong to this worker.
  4. When a worker rejects a task, Temporal retries it. Eventually the right worker (the one whose routing key set includes this task's key) pulls and processes it.
  5. Activity interceptors bridge the routing key from task headers into OpenTelemetry baggage for the duration of activity execution, so outbound HTTP calls made from activities automatically carry the key and route correctly downstream.

The result is an invariant: exactly one worker version processes each task. A task with no routing key belongs to the baseline. If a routing key is no longer claimed by a sandbox, the task also falls back to the baseline. A sandbox that still exists continues to claim its key even when its worker is unavailable, so those tasks wait or retry instead of falling back.

What you will deploy​

The demo app deploys to the temporal namespace:

  • py_client (FastAPI web UI, shared): starts workflows with OpenTelemetry context propagation, works with every worker via its TASK_QUEUE setting
  • Temporal server (one instance, all languages): Temporal with embedded Postgres and Elasticsearch
  • One worker in your language: temporal_worker (Python), ts_worker (TypeScript), go_worker (Go), or java_worker (Java); the baseline and its sandboxed variants share that language's task queue
  • K8s manifests: the Temporal server stack plus baseline Deployments for the workers and client; Sandbox specs for all four languages live in sandbox/

Deploy the demo​

Prerequisites:

  • A Minikube cluster with the Signadot Operator installed
  • The signadot CLI installed and authenticated
  • kubectl pointing at the cluster
  • Docker and the minikube CLI installed

Build the shared web client image and load it into Minikube:

docker build -t temporal-py-client-ui:v1.0 py_client
minikube image load temporal-py-client-ui:v1.0

Deploy the Temporal server stack (server, Postgres, Elasticsearch, UI, admin tools) to the temporal namespace:

kubectl create namespace temporal
kubectl apply -n temporal -f k8s/temporal/

Then deploy the baseline worker and the web client:

docker build -t temporal-money-transfer:v1.0 temporal_worker
minikube image load temporal-money-transfer:v1.0

kubectl apply -n temporal -f k8s/worker-deployment.yaml
kubectl apply -n temporal -f k8s/temporal-py-client-ui-deployment.yaml

The Python worker polls the money-transfer task queue.

Wait for all pods to be Ready:

kubectl get pods -n temporal -w

How the code works​

The pattern is platform-owned isolation logic (interceptors, routeserver client) plus ordinary application code (workflows, activities) that never touches Signadot. The integration points:

1. The routing-key gate in workflow interceptors​

When a worker receives a workflow task, the interceptor extracts the routing key from the task headers (stamped by the client's OpenTelemetry tracer), checks the Routes API to see if this task belongs to this worker, and rejects it if not.

The workflow interceptor reads baggage directly because the SDK's TracingInterceptor has already attached it:

temporal_worker/signadot/interceptors.py
            async def execute_workflow(self, input: ExecuteWorkflowInput):
                # The TracingInterceptor (registered before this interceptor) has
                # already attached the OTel context from the workflow headers, so
                # baggage is readable directly here.
                routing_key = str(baggage.get_baggage(ROUTING_KEY) or "")
                workflow_name = getattr(input.run_fn, "__name__", str(input.run_fn))
                should_process = True

                if self.routes_client and not await self.routes_client.should_process(routing_key):
                    should_process = False

                if not should_process:
                    error_msg = f"Workflow/Worker cannot handle routing key: {routing_key} - Worker: {self.worker_ident}"
                    logger.info(error_msg)
                    raise Exception(error_msg)

                logger.info(f"[Worker:{self.worker_ident}] Workflow: {workflow_name}: Processing task with routing key '{routing_key}'")
                return await self.next.execute_workflow(input)

Note what is NOT here: no sandbox awareness in the application. Workflows contain pure domain logic (withdraw, deposit); the routing check is owned by the platform layer.

2. Learn which routing keys are claimed​

Workers need to know which keys currently belong to live sandboxes. That state lives in the Signadot Routes API, served in-cluster:

curl "http://routeserver.signadot.svc:7778/api/v1/workloads/routing-rules?baselineKind=Deployment&baselineNamespace=temporal&baselineName=temporal-worker"
# {"routingRules": [{"routingKey": "k7", ...}]}

A sandboxed worker adds &destinationSandboxName=<its-sandbox> so it receives only the keys it should claim. The poll-and-cache client maintains a refreshed cache of routing keys:

temporal_worker/signadot/routing.py
class RoutesAPIClient:
    """
    Client for fetching routing rules from the central platform routing API.
    Handles caching and refresh of routing keys for sandbox/baseline selection.
    """
    def __init__(self, sandbox_name: str):
        self.sandbox_name = sandbox_name
        self.route_server_addr_base = os.environ["ROUTES_API_ROUTE_SERVER_ADDR"]
        parsed_addr = urlparse(self.route_server_addr_base)
        self.route_server_scheme = parsed_addr.scheme or "http"
        self.route_server_netloc = parsed_addr.netloc
        self.baseline_kind = os.environ["ROUTES_API_BASELINE_KIND"]
        self.baseline_namespace = os.environ["ROUTES_API_BASELINE_NAMESPACE"]
        self.baseline_name = os.environ["ROUTES_API_BASELINE_NAME"]
        self.refresh_interval = int(os.environ["ROUTES_API_REFRESH_INTERVAL_SECONDS"])
        self._routing_keys_cache: Set[str] = set()

3. Gate the activity before processing​

When a worker receives an activity task, the interceptor applies the same routing check as workflows:

temporal_worker/signadot/interceptors.py
            async def execute_activity(self, input: ExecuteActivityInput):
                # Unlike workflows, the SDK's TracingInterceptor only uses the
                # context from the headers to parent the activity span -- it does
                # NOT attach it, so baggage is not available via get_baggage()
                # here. Extract it from the headers explicitly instead.
                header_context = context_from_headers(input.headers)
                routing_key = routing_key_from_context(header_context)
                activity_name = getattr(input.fn, "__name__", str(input.fn))
                should_process = True

                if self.routes_client and not await self.routes_client.should_process(routing_key):
                    should_process = False

                if not should_process:
                    error_msg = f"Activity/Worker cannot handle routing key: {routing_key} - Worker: {self.worker_ident}"
                    logger.info(error_msg)
                    raise Exception(error_msg)

                logger.info(f"[Worker:{self.worker_ident}] Activity: {activity_name}: Processing task with routing key '{routing_key}'")

                # Bridge baggage from the task headers into the current OTel
                # context, scoped to this activity execution. Without this,
                # outbound HTTP calls made by the activity would carry the trace
                # context but NOT the sd-routing-key baggage, so downstream
                # services would not route sandbox traffic correctly.
                context = otel_context.get_current()
                if header_context is not None:
                    for key, value in baggage.get_all(header_context).items():
                        context = baggage.set_baggage(key, value, context=context)
                token = otel_context.attach(context)
                try:
                    logger.info(
                        f"[Worker:{self.worker_ident}] Activity: {activity_name}: outbound HTTP calls will carry: {outbound_http_headers()}"
                    )
                    return await self.next.execute_activity(input)
                finally:
                    otel_context.detach(token)

This is the critical step: without bridging baggage into the OTel context, outbound HTTP calls made from activities would carry traceparent but no baggage: sd-routing-key=..., and downstream sandbox routing would silently break. The worker logs the effective outbound headers per activity so this is easy to verify.

4. Worker setup and the sandbox-aware wrapper​

The entry point registers interceptors, polls the routeserver, and starts the worker. Application developers never touch this:

temporal_worker/main.py
async def main():
    # Get task queue from environment
    task_queue = os.environ["TASK_QUEUE"]

    # Create banking activities instance
    banking_activities = BankingActivities()

    # Create the SandboxAware worker
    worker = SandboxAwareWorker(
        task_queue=task_queue,
        workflows=[MoneyTransferWorkflow],
        activities=[
            banking_activities.withdraw,
            banking_activities.deposit,
        ]
    )

    # Start the worker
    await worker.start()

SandboxAwareWorker wires up the interceptors, starts the routeserver poller, and instruments aiohttp so outbound HTTP calls carry the routing key automatically.

5. Sandbox spec​

Each sandboxed worker is a fork of the baseline Deployment:

sandbox/worker-sandbox.yaml
# see https://www.signadot.com/docs/reference/sandboxes/spec
name: temporal-worker-sandbox
spec:
  labels:
    team: backend
  cluster: "@{cluster}"
  description: Testing temporal sandboxes
  defaultRouteGroup:
    endpoints:
    - name: web-client
      target: http://temporal-py-client-ui.temporal.svc:8080
  forks:
  - forkOf:
      kind: Deployment
      namespace: temporal
      name: temporal-worker
    customizations:
      images:
      - container: temporal-worker
        image: "@{image}"

Create sandboxes and test​

Create a sandbox for the worker language you deployed above:

Build the sandbox worker image and load it into Minikube:

docker build -t temporal-money-transfer:sandbox temporal_worker
minikube image load temporal-money-transfer:sandbox

signadot sandbox apply -f sandbox/worker-sandbox.yaml \
  --set cluster=<your-cluster-name> \
  --set image=temporal-money-transfer:sandbox

The command prints the sandbox's routing key and a preview URL for the web UI. Opening the preview URL tags your requests with that sandbox's routing key automatically.

Open the preview URL, select two accounts, enter an amount, and start the workflow. The success panel shows the workflow ID and confirms that the web client received the sandbox routing key:

Web client confirming a tagged money-transfer workflow submission

Alternatively, select the values for your worker language:

SANDBOX_NAME=temporal-worker-sandbox
BASELINE_DEPLOYMENT=temporal-worker
WORKER_CONTAINER=temporal-worker

Extract the routing key:

ROUTING_KEY=$(signadot sandbox get "$SANDBOX_NAME" -o json | jq -r '.routingKey')

In a separate terminal, forward the web client port:

kubectl port-forward service/temporal-py-client-ui -n temporal 8080:8080

Return to the first terminal. Submit one tagged workflow and one untagged workflow. The endpoint accepts form fields, not JSON:

curl -X POST http://localhost:8080/api/start-workflow \
  -H "baggage: sd-routing-key=$ROUTING_KEY" \
  -d "from_account=acc_001" -d "to_account=acc_002" -d "amount=100.00"

curl -X POST http://localhost:8080/api/start-workflow \
  -d "from_account=acc_001" -d "to_account=acc_002" -d "amount=100.00"

After submitting one tagged and one untagged workflow, both runs should complete:

Temporal UI showing completed tagged and untagged money-transfer workflows

Check the worker logs to see which version processed the task:

kubectl logs -n temporal "deployment/$BASELINE_DEPLOYMENT" \
  -c "$WORKER_CONTAINER" --since=10m | \
  grep "Processing task with routing key"

SANDBOX_DEPLOYMENT=$(kubectl get deployments -n temporal -o name | \
  grep "^deployment.apps/${SANDBOX_NAME}-dep-${BASELINE_DEPLOYMENT}-")

kubectl logs -n temporal "$SANDBOX_DEPLOYMENT" \
  -c "$WORKER_CONTAINER" --since=10m | \
  grep "Processing task with routing key"

You should see the message:

Processing task with routing key '<ROUTING_KEY>'

The baseline worker processes tasks without a routing key, while the sandbox worker processes tasks carrying the sandbox routing key:

Baseline and sandbox worker logs showing task processing by routing key

Three scenarios to verify:

Routing contextSource workerProcessing workerWhy
none (baseline)baselinebaselineuntagged workflows belong to the baseline
sandboxedsandboxedsandboxedthe routing key is claimed; the baseline skips it
unclaimed keyany clientbaselinethe Routes API has no sandbox rule for this key, so fallback applies

The third row is the fallback rule: if a workflow carries a key that is not claimed by any sandbox rule, the baseline processes it. A sandbox that still exists but has an unavailable worker continues to claim its key, so its tasks wait or retry rather than falling back.

Behavior and tradeoffs​

  • Message-based isolation is recommended. Shared task queues with routing keys in task headers and per-task routing via the Routes API work well with Temporal's task distribution model. Per-sandbox task queues are NOT recommended: they push routing decisions onto every workflow starter, lose baseline fallback for unforked services, and do not compose with Route Groups.

  • No first-pickup guarantee. The pattern relies on wrong-worker rejections and server retries. A task may bounce between workers before the right one claims it. This is acceptable for integration testing because test tasks are short-lived and bounded.

  • Retry exhaustion. Wrong-worker rejections can burn bounded retry budgets. The workflow interceptor must use the SDK-specific failure mechanism that fails only the current workflow task, rather than failing the workflow execution. The activity interceptor should return a retryable failure with a short retry delay, about 1 second, so routing bounces skip exponential backoff. Keep activity attempts unbounded within a schedule-to-close timeout when the SDK and application policy allow it. If attempts must be bounded, size the limit for the expected number of wrong-worker bounces.

  • Coupled services and Route Groups. When a workflow spans services that must be tested together, create one sandbox with multiple forked workloads (they share a routing key). Route Groups also work, but only if workers consult the Routes API as the source of truth, not the SIGNADOT_SANDBOX_NAME environment variable. Env vars are immutable and never reflect Route Group keys. This example queries the Routes API directly, so Route Groups work transparently.

  • Advanced alternative: producer-side queue routing. To avoid retries entirely, a producer-side interceptor can consult the Routes API at workflow-submit time and start the work on a task queue named <queue>--<routing-key>. Sandboxed workers poll only that queue. Caveats: the baseline producer needs the interceptor too; inside workflows the lookup must run in a local activity or side effect for replay determinism; tasks already on <queue>--<rk> strand (until timeout) if the sandbox is deleted mid-flight instead of falling back to baseline. The reject-and-retry approach (this tutorial) is the more common choice because it is simpler and the baseline fallback is more robust.

Adapt this to your stack​

The reference platform modules live in temporal_worker/signadot/ (Python), ts_worker/src/signadot/ (TypeScript), go_worker/signadot/ (Go), and java_worker/src/main/java/com/signadot/temporaldemo/signadot/ (Java). Use the implementation for your SDK as the starting point:

  1. Add the platform layer to your shared code:

    • workflow and activity interceptors: the routing gate and baggage bridge
    • a Routes API client with a polling cache
    • a worker wrapper or factory that wires the platform layer into the Temporal SDK

  2. Instrument your HTTP clients. Use the matching OpenTelemetry instrumentation package for the HTTP client your activities call. The Python and TypeScript implementations also include helpers for injecting the current baggage into outbound headers when automatic instrumentation is unavailable.

  3. Set environment variables on your worker Deployment:

    TEMPORAL_SERVER_URL
    TASK_QUEUE
    ROUTES_API_ROUTE_SERVER_ADDR
    ROUTES_API_BASELINE_KIND
    ROUTES_API_BASELINE_NAMESPACE
    ROUTES_API_BASELINE_NAME
    ROUTES_API_REFRESH_INTERVAL_SECONDS

    SIGNADOT_SANDBOX_NAME is injected automatically by the operator.

  4. Adjust retry policies. Size maximum_attempts to account for routing bounces (default is unbounded; recommend keeping it unbounded for sandboxed work).

  5. Register your workflows and activities through the language's sandbox-aware worker wrapper or factory. Your application workflows and activities remain separate from the routing implementation.

Cleanup​

Delete the sandbox for the worker language you tested:

signadot sandbox delete temporal-worker-sandbox

Then delete the demo namespace:

kubectl delete namespace temporal

See also​