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]

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, 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).