Tenant-sharded realtime

The default realtime stream is tenant-global: one stream carries every tenant's signals and the tenant rides an (untrusted) header. For trusted per-tenant isolation, put the stream on the tenancy tier ladder — wire the stream tenant_aware so each tenant gets its own key, and consume one loop per tenant so a signal's tenant is the stream it came from, never a forgeable header. The Socket.IO gateway automates the consume side with TenantShardedSignalSource; this recipe shows what makes it work end to end for durable signals.

Wire the stream tenant-aware, keep the outbox global¶

Wire only the realtime stream route tenant_aware (each tenant gets its own key). The outbox stays tenant-global — a shared table whose rows are tagged with their tenant — so the tenant-less relay can drain every tenant from one place:

ACME = UUID("11111111-1111-1111-1111-111111111111")  # tenant A
GLOBEX = UUID("22222222-2222-2222-2222-222222222222")  # tenant B

STREAM = realtime_stream_spec()  # the per-tenant stream (wired tenant_aware below)
# The outbox keeps its own name so it can stay tenant-global while the stream is tenant-aware
# (the mock keys routes by spec name; in production they are separate backends).
OUTBOX = realtime_outbox_spec(name="realtime-outbox", stream=str(STREAM.name))


class _OrderShipped(BaseModel):
    order: str


SHIPPED = RealtimeEvent(name="order.shipped", payload_type=_OrderShipped)


def _context() -> ExecutionContext:
    routes = {
        str(STREAM.name): MockRouteConfig(tenant_aware=True),  # per-tenant stream key
        # "realtime-outbox" is intentionally absent → tenant-global (tagged) outbox
    }
    return ExecutionContext(
        deps=DepsRegistry.from_modules(MockDepsModule(routes=routes)).freeze().resolve()
    )

Stage a durable signal under its tenant¶

A handler stages the signal with no realtime or tenant plumbing — the staging tenant is ambient, and the outbox row is tagged with it:

async def publish_durable(ctx: ExecutionContext, *, tenant: UUID, order: str) -> None:
    """A handler stages a durable signal under its tenant — no realtime/tenant plumbing.

    The signal is addressed to a principal; the staging tenant is ambient. The outbox row
    is tagged with that tenant, which the relay later uses to route it. (A real handler
    flushes in its own transaction; here both tenants stage into one buffer, flushed once
    by :func:`relay` so the in-process demo shares a single mock store.)
    """

    with ctx.inv_ctx.bind_identity(tenant=TenantIdentity(tenant_id=tenant)):
        publisher = build_realtime_publisher(
            ctx, stream_spec=STREAM, outbox_spec=OUTBOX
        )
        await publisher.stage(
            Audience.principal("alice"), SHIPPED, _OrderShipped(order=order)
        )

The relay routes each row to its tenant's stream¶

The relay runs with no tenant bound, yet routes correctly: it binds each row's staged tenant before appending, so a durable signal lands on that tenant's stream key.

async def relay(ctx: ExecutionContext) -> None:
    """The background relay drains the (tenant-global) outbox and forwards each row.

    It runs with **no** tenant bound, yet routes correctly: it binds each row's staged
    tenant before appending, so the durable signal lands on that tenant's stream key.
    """

    await ctx.outbox.command(
        OUTBOX
    ).flush()  # write staged rows to the tenant-global outbox
    await OutboxRelay(outbox_spec=OUTBOX).to_stream(ctx, STREAM)

Consume one tenant's stream¶

Binding a tenant resolves the stream adapter to that tenant's key, so a per-tenant consumer only ever sees its own signals — the isolation is the stream's, not a header check. This is what TenantShardedSignalSource runs once per assigned tenant:

async def read_tenant_stream(
    ctx: ExecutionContext, tenant: UUID
) -> list[RealtimeSignal]:
    """What ``TenantShardedSignalSource`` does per assigned tenant: bind it, read its stream.

    Binding the tenant resolves the stream adapter to *that tenant's* key, so this only ever
    sees the bound tenant's signals — the isolation is the stream's, not a header check.
    """

    with ctx.inv_ctx.bind_identity(tenant=TenantIdentity(tenant_id=tenant)):
        stream = ctx.deps.resolve_configurable(
            ctx, StreamQueryDepKey, STREAM, route=STREAM.name
        )
        messages = await stream.read({str(STREAM.name): "0"})

    return [m.payload for m in messages]

Assign tenants at startup¶

The shard is a fixed snapshot, so something has to pick each replica's tenants before its gateway starts. Fleet membership can: each replica beats as starting through one ttl, so replicas starting together see each other, then takes the tenants it ranks first for among the serving and starting members:

GATEWAYS = FleetSpec(name="realtime-gateways", ttl=timedelta(seconds=10))


async def assign_shard(
    fleet: FleetPort,
    me: str,
    tenants: Sequence[UUID],
    *,
    spec: FleetSpec = GATEWAYS,
) -> RealtimeShard:
    """Take the tenants this replica ranks first for, among replicas serving or starting."""

    # Beat through one ttl, at a third of it, so replicas starting meanwhile see this one
    # and it sees them; rank over the last beat's view.
    view = await fleet.beat(me, "starting", {}, ttl=spec.ttl)

    for _ in range(3):
        await asyncio.sleep(spec.ttl.total_seconds() / 3)
        view = await fleet.beat(me, "starting", {}, ttl=spec.ttl)

    mine = [
        t for t in tenants if view.rank(str(t), states=("serving", "starting"))[0] == me
    ]

    return RealtimeShard(stream_spec=STREAM, tenants=mine)

The shard is consumed when the steps are built, before the runtime exists, so build the port directly — a RedisFleetAdapter over a RedisClient you initialize for it. Once assigned, keep the replica in the fleet with the membership step on the same FleetSpec and member id, so replicas started later rank it in; it beats serving once startup finishes. The first replica of a fresh fleet finds only itself and takes every tenant.

The assignment is a snapshot, sound only when every replica starts within one ttl of the others — an all-at-once rollout — and only while the fleet keeps that shape:

a rolling deploy or restart leaves tenants with no owner: each new replica leaves to an old one the tenants it ranks first for, and the old one is then stopped;
a replica that dies leaves its tenants unowned until the fleet restarts;
a replica that starts later, or whose beats land after another's read, can take tenants another replica also holds, or leave some that no replica holds.

Moving tenants between running replicas is not covered; to reassign, restart every replica at once.

Notes¶

The outbox stays tenant-global by design. Only the stream is tenant_aware; the shared outbox is drained by the standard relay with per-row tenant routing. To partition the outbox table itself, wire its route tenant_aware too and use the sharded relay (realtime_tenant_relay_lifecycle_step).
In production, TenantShardedSignalSource(shard=…) replaces the hand-written per-tenant read: one consume loop per assigned tenant, bound to it. Hand the same RealtimeShard to the source, the group-ensure step, and (for a partitioned outbox) the relay so they can't drift.
Assignment, not discovery — the shard is a fixed snapshot resolved at startup (by rank, as above, or by your own assignment), so onboarding a new tenant (or rebalancing) needs a restart. Broker-level enforcement (so a rogue producer can't write another tenant's key) is the operator's job (Redis ACLs).