Propagate sessions

Before writing tests, set up x-varianz-id propagation in each instrumented service. This is a one-time setup per service; afterwards, every VPoint you add works with test sessions automatically. The session ID must be:

Extracted from incoming requests (HTTP header or gRPC metadata)
Held in the execution context for the duration of the request
Injected into all outgoing requests to downstream services

Step 3 is the one teams miss. If a service calls downstream services without forwarding the ID, downstream VPoints silently never see your test’s stages.

At your API gateway, strip or override x-varianz-id (and baggage) headers arriving from the public internet. Only test infrastructure should set them.

With OpenTelemetry Baggage

If your services already run OTel with the W3C Baggage propagator (most default setups include it), propagation across HTTP and gRPC hops is nearly free — carry x-varianz-id as a baggage entry and OTel’s instrumented clients and servers move it for you.

from opentelemetry import baggage

# Set at the entry point:
context = baggage.set_baggage("x-varianz-id", session_hex)

# Read anywhere downstream:
session_id = baggage.get_baggage("x-varianz-id")

import { propagation, context } from '@opentelemetry/api';

const bag = propagation.getBaggage(context.active());
const sessionId = bag?.getEntry('x-varianz-id')?.value;

import "go.opentelemetry.io/otel/baggage"

func getSessionID(ctx context.Context) string {
    return baggage.FromContext(ctx).Member("x-varianz-id").Value()
}

import io.opentelemetry.api.baggage.Baggage;

// Read:
String sessionId = Baggage.current().getEntryValue("x-varianz-id");

// Set (downstream calls in scope auto-propagate):
try (Scope scope = Baggage.builder().put("x-varianz-id", sessionHex).build().makeCurrent()) {
    // instrumented HTTP/gRPC calls carry it automatically
}

You still need to hand the extracted ID to the Varianz SDK (Varianz.set_session, varianz.WithSession, etc.) — the sections below show where.

Python

A gRPC server interceptor extracts the ID and activates session routing; handlers need no per-request code:

import grpc
from varianz import Varianz

class VarianzInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        session_id = metadata.get("x-varianz-id")
        if session_id:
            Varianz.set_session(session_id)
        return continuation(handler_call_details)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[VarianzInterceptor()],
)

For HTTP, the same two lines go in middleware — @app.before_request in Flask, or an ASGI middleware in FastAPI that reads the x-varianz-id header and calls Varianz.set_session(...). Outbound: forward the current session on downstream calls.

def _make_metadata():
    session_id = Varianz.get_current_session()
    return [("x-varianz-id", session_id)] if session_id else []

response = stub.SomeMethod(request, metadata=_make_metadata())

The session is held in a contextvar, so it copies across asyncio tasks automatically. Varianz and OTel interceptors coexist without conflict.

Go

Set the session on the request context.Context with varianz.WithSession; both varianz.VPoint callables and proxy.Call read it from there.

// gRPC server interceptor — extract, activate, and forward in one place.
func varianzUnaryInterceptor(
    ctx context.Context, req interface{},
    info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (interface{}, error) {
    if md, ok := metadata.FromIncomingContext(ctx); ok {
        if ids := md.Get("x-varianz-id"); len(ids) > 0 {
            ctx = varianz.WithSession(ctx, ids[0])
            ctx = metadata.AppendToOutgoingContext(ctx, "x-varianz-id", ids[0])
        }
    }
    return handler(ctx, req)
}

server := grpc.NewServer(grpc.ChainUnaryInterceptor(varianzUnaryInterceptor))

// HTTP middleware (stdlib/chi/gorilla; adapt for Gin etc.)
func VarianzMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if sid := r.Header.Get("x-varianz-id"); sid != "" {
            r = r.WithContext(varianz.WithSession(r.Context(), sid))
        }
        next.ServeHTTP(w, r)
    })
}

// Outbound HTTP/gRPC: read it back when building the request.
if sid := varianz.SessionFromContext(ctx); sid != "" {
    ctx = metadata.AppendToOutgoingContext(ctx, "x-varianz-id", sid)
}

Never spawn a goroutine without passing the parent context.Context — the session ID is lost with it.

TypeScript

The SDK holds the session in AsyncLocalStorage, exposed as VarianzContext. It survives await, timers, and event emitters.

import { VarianzContext } from '@varianz/node';

// Express / Connect: one line.
app.use(VarianzContext.expressMiddleware());

// Any other entry point: enter the scope explicitly.
await VarianzContext.session(sessionHex, async () => {
  await service.handleRequest(input);
});

// Outbound: attach the registered headers to downstream calls.
await fetch(downstreamUrl, { headers: VarianzContext.getHeaders() });

For Fastify, Koa, and Hono, read the header in an onRequest/middleware hook and wrap the rest of the request in VarianzContext.session(...). For NestJS, nestjs-cls provides the same AsyncLocalStorage plumbing. gRPC (@grpc/grpc-js): server interceptors return synchronously before the handler runs, so they can’t wrap the async handler in the storage scope. Bind the session inside the handler instead:

function charge(call: ServerUnaryCall<Req, Resp>, callback: sendUnaryData<Resp>) {
  const sid = call.metadata.get('x-varianz-id')[0] as string | undefined;
  const run = () => doCharge(call.request, callback);
  sid ? VarianzContext.session(sid, run) : run();
}

If a method already receives the session as a parameter, declare it with varianzId('paramName') in the @vpoint params and the SDK enters the scope automatically.

Java and Kotlin

The JVM SDK pulls the session ID on every VPoint invocation through a single static method you provide, annotated @VarianzId @Function(global = true). Your framework interceptor stashes the incoming header in a request-scoped carrier; the annotated method reads it back. This one pattern works for gRPC, Spring MVC, WebFlux, and virtual threads — only the carrier changes. Step 1 — the binding class (ThreadLocal shown; see the carrier table below):

import io.varianz.annotations.struct.Function;
import io.varianz.annotations.vpoint.param.VarianzId;

public final class VarianzSessionBinding {
    private static final ThreadLocal<String> CURRENT = new ThreadLocal<>();

    public static void set(String sessionId) { CURRENT.set(sessionId); }
    public static void clear()               { CURRENT.remove(); }

    @VarianzId
    @Function(name = "sessionOverride", global = true)
    public static String sessionOverride() {
        return CURRENT.get();
    }
}

Step 2 — stash the header in an interceptor:

// gRPC
public class VarianzServerInterceptor implements ServerInterceptor {
    private static final Metadata.Key<String> VARIANZ_ID =
        Metadata.Key.of("x-varianz-id", Metadata.ASCII_STRING_MARSHALLER);

    @Override
    public <Req, Resp> ServerCall.Listener<Req> interceptCall(
            ServerCall<Req, Resp> call, Metadata headers, ServerCallHandler<Req, Resp> next) {
        String sessionId = headers.get(VARIANZ_ID);
        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<>(
                next.startCall(call, headers)) {
            @Override
            public void onMessage(Req message) {
                VarianzSessionBinding.set(sessionId);
                try { super.onMessage(message); }
                finally { VarianzSessionBinding.clear(); }
            }
        };
    }
}

For Spring MVC, do the same in a HandlerInterceptor (preHandle sets, afterCompletion clears) and register it via WebMvcConfigurer. Step 3 — forward downstream:

Metadata headers = new Metadata();
String sid = VarianzSessionBinding.sessionOverride();
if (sid != null) headers.put(VARIANZ_ID, sid);
stub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers))
    .someMethod(request);

For HTTP clients (RestTemplate, WebClient, OkHttp), an outbound interceptor reads sessionOverride() and sets the header the same way. Pick the carrier for your concurrency model. Plain ThreadLocal is correct for thread-per-request servlet and gRPC work, and wrong for reactive code:

Framework	Carrier
Spring MVC / servlet (sync)	`ThreadLocal`
Spring WebFlux (reactive)	Reactor `Context`, mirrored to a ThreadLocal via Micrometer `ContextRegistry` + `Hooks.enableAutomaticContextPropagation()`
Spring `@Async` / executors	ThreadLocal + a `TaskDecorator` that copies it
Virtual threads (JDK 21+)	`ScopedValue`

The annotated method runs synchronously on the VPoint caller thread, so it must read from a carrier that’s populated on that thread — don’t try to block() on a reactive context inside it.

Services with explicit header-forwarding lists

Some codebases forward a fixed set of headers at the application level (a getForwardHeaders()-style function listing x-request-id, x-b3-traceid, …) instead of using interceptors. If yours does, add x-varianz-id to that list — a one-line change that carries the session through the whole chain:

FORWARDED_HEADERS = ["x-request-id", "x-b3-traceid", "x-b3-spanid", "x-varianz-id"]

To find these, search for patterns like forward.*header, getForward, or existing tracing headers in the codebase.

Verify before writing tests

Start a local registry and your services.

Send a request with a fake session ID:

curl -H "x-varianz-id: 0123456789abcdef0123456789abcdef" http://your-service/api

Check each downstream service’s logs to confirm the ID arrived at every hop.

A stage that never applies in a multi-service test almost always means one hop in this chain is missing.

Getting started

Core concepts

Guides

SDK reference

Testing

Reference

Propagate sessions

With OpenTelemetry Baggage

Python

Go

TypeScript

Java and Kotlin

Services with explicit header-forwarding lists

Verify before writing tests

​With OpenTelemetry Baggage

​Python

​Go

​TypeScript

​Java and Kotlin

​Services with explicit header-forwarding lists

​Verify before writing tests

With OpenTelemetry Baggage

Python

Go

TypeScript

Java and Kotlin

Services with explicit header-forwarding lists

Verify before writing tests