Observable query emission guards

Authorization for an observable query runs when the subscription is established. That verdict decides whether the caller may obtain the live stream — it says nothing about the minutes or hours the stream then stays open.

That is fine for most applications. Subscriptions end when the user navigates away, the tab closes, or the server shuts down. But a subscription has no other reason to end: nothing terminates it when a token expires, a session is signed out, or a role is revoked. Keep-alive pings deliberately stop proxies from culling an idle connection, and the client reconnects and re-subscribes on its own.

An emission guard closes that gap. Implement IGuardObservableQueryEmission and Arc consults it for every emission an observable query is about to write, on every transport — the multiplexed hub, direct WebSocket/SSE connections, and plain-HTTP observable snapshots alike. Guards also check non-streaming query results delivered through the hub. Read-model interception runs before the guard.

Writing a guard

Illustrative application integration fragment: ISessions is your application’s real session store contract, not an Arc API. Supply and register an implementation that checks authoritative session state; do not replace it with an always-true stub. This guard deliberately requires a session for every stream or hub snapshot it evaluates.

using System.Threading.Tasks;
using Cratis.Arc.Queries;

namespace Application.Security;

public interface ISessions
{
    Task<bool> IsActive(string sessionId);
}

public class SessionMustStillBeActive(ISessions sessions) : IGuardObservableQueryEmission
{
    public async Task<ObservableQueryEmissionVerdict> Guard(ObservableQueryEmissionContext context)
    {
        var sessionId = context.Principal?.FindFirst("sid")?.Value;
        if (sessionId is null)
        {
            return ObservableQueryEmissionVerdict.DenyAndTerminate;
        }

        return await sessions.IsActive(sessionId)
            ? ObservableQueryEmissionVerdict.Allow
            : ObservableQueryEmissionVerdict.DenyAndTerminate;
    }
}

That is the whole opt-in. Guards are discovered by convention — no registration, no configuration. An application with no guard builds no emission context and dispatches nothing; emissions take their existing path. A filter that supplies a scope still incurs the subscription-time snapshot, even if no guard is installed.

Compare the scope captured at subscription time

When a query filter chooses which organization’s data to stream, its decision may change while the connection stays open. Put a serializable value in QueryContext.SubscriptionScope during filtering in Arc’s query pipeline. Arc captures it after the filters succeed and before the query executes, then passes a fresh copy in ObservableQueryEmissionContext.SubscriptionScope on each emission. Values assigned during query execution are too late and do not change the captured scope. The guard can compare that original scope to current membership without rerunning the filter or changing the query arguments.

Controller-based observable queries do not run IQueryFilters through QueryActionFilter.EstablishQueryContext. Their emission guards always receive a null subscription scope; use a query handled by Arc’s query pipeline if you need a filter-supplied scope.

This application integration fragment uses IMemberships as an application-owned contract. Supply an implementation backed by authoritative membership state; CurrentOrganizationId must return the current organization, not a value cached at login. Replace the query name with your observable query’s fully qualified name. The query itself must use the same filter-derived organization to select its data — setting the scope alone does not filter results.

using System.Security.Claims;
using System.Threading;
using System.Threading.Tasks;
using Cratis.Arc.Authorization;
using Cratis.Arc.Queries;

namespace Application.Security;

public interface IMemberships
{
    Task<string?> CurrentOrganizationId(ClaimsPrincipal principal, CancellationToken cancellationToken);
}

public class CaptureOrganizationScope(ICurrentPrincipalAccessor principalAccessor, IMemberships memberships) : IQueryFilter
{
    public async Task<QueryResult> OnPerform(QueryContext context)
    {
        if (context.Name.Value != "Application.Queries.ForCurrentOrganization")
        {
            return QueryResult.Success(context.CorrelationId);
        }

        var principal = principalAccessor.Current;
        if (principal is null)
        {
            return QueryResult.Unauthorized(context.CorrelationId);
        }

        var organizationId = await memberships.CurrentOrganizationId(principal, context.CancellationToken);
        if (organizationId is null)
        {
            return QueryResult.Unauthorized(context.CorrelationId);
        }

        context.SubscriptionScope = organizationId;
        return QueryResult.Success(context.CorrelationId);
    }
}

public class OrganizationMustStillMatch(IMemberships memberships) : IGuardObservableQueryEmission
{
    public async Task<ObservableQueryEmissionVerdict> Guard(ObservableQueryEmissionContext context)
    {
        if (context.QueryName.Value != "Application.Queries.ForCurrentOrganization")
        {
            return ObservableQueryEmissionVerdict.Allow;
        }

        if (context.Principal is null || context.SubscriptionScope is not string originalOrganization)
        {
            return ObservableQueryEmissionVerdict.DenyAndTerminate;
        }

        var currentOrganization = await memberships.CurrentOrganizationId(context.Principal, context.CancellationToken);
        return currentOrganization == originalOrganization
            ? ObservableQueryEmissionVerdict.Allow
            : ObservableQueryEmissionVerdict.DenyAndTerminate;
    }
}

For example, a subscription established while membership is former captures former. If membership changes to new with unchanged query arguments, the next emission still carries former, and the guard returns DenyAndTerminate. A subscription created after the change captures new independently. With no scope supplied, the emission context’s scope is null; a supplied scope that cannot be serialized and restored fails the query explicitly as InvalidSubscriptionScope rather than silently becoming null. Keep the scope small: prefer primitives or records of primitive values. Only data serialized by Arc’s JsonSerializerOptions survives the snapshot. Private fields, [JsonIgnore] members, and derived members of non-polymorphic base-typed properties are dropped even when the remaining value can round-trip. Do not put live services or mutable authorization caches in it.

The three verdicts

Verdict	What happens
`Allow`	The emission is written unchanged.
`Suppress`	This one emission is withheld. The subscription stays live and the next emission is evaluated again.
`DenyAndTerminate`	Nothing is written, the client is told it is unauthorized, and this subscription is torn down.

For an HTTP observable snapshot, Allow returns intercepted data with 200 OK and DenyAndTerminate returns 403 Forbidden without data. Suppress returns 202 Accepted without data if you did not request waiting. With waitForFirstResult=true, Arc keeps evaluating emissions until one is allowed, the query is denied, or the original deadline expires with 408 Request Timeout. Suppression does not restart the deadline; every candidate for this snapshot carries IsFirstEmission=true.

For a non-streaming query delivered through the hub, the guard receives IsSnapshot=true and IsFirstEmission=true. Allow sends the one-shot result; DenyAndTerminate sends Unauthorized without data (and SSE subscribe returns 401). Suppress sends no result, error, or denial frame and ends the operation: there is no stream to evaluate again. The SSE subscribe POST returns 200 for suppression, not an indication that data arrived.

Suppress does not move the delta baseline. The client never saw the withheld emission, so the next delivered ChangeSet is still computed against the last state it actually received — nothing goes missing.

DenyAndTerminate ends only the subscription it was given. Every sibling subscription on the same multiplexed connection keeps streaming. On the hub the client receives an Unauthorized message for that query id and the subscription is deleted; on a direct connection it receives a final QueryResult with IsAuthorized false and the stream closes. In both cases the Arc client latches the denial and stops reconnecting — a reconnect would only be denied again, and on a direct SSE connection the browser would otherwise re-establish the stream every few seconds and re-run the whole query for each attempt.

What the guard is told

ObservableQueryEmissionContext carries the fully qualified query name, the coerced query arguments, the captured SubscriptionScope (or null), the caller’s ClaimsPrincipal, the correlation id, whether this is the first emission on the subscription, IsSnapshot (true only for non-streaming hub results; false for stream emissions and HTTP observable snapshots), the subscription’s cancellation token, and the IServiceProvider to resolve from. Each guard receives its own copy of the scope and arguments; mutation by one guard cannot change another guard’s input or a later emission.

The principal is handed over explicitly. Emissions arrive on the producing stream’s own thread, where the request’s AsyncLocal context does not flow, so a guard that reached for an ambient accessor would see the wrong identity — or none at all.

Several guards

Every guard is asked, and the most restrictive verdict wins: DenyAndTerminate over Suppress over Allow. The first DenyAndTerminate short-circuits the rest.

Failing closed

A guard that throws is treated as DenyAndTerminate. The failure is logged at Error and never rethrown.

This is deliberate. A guard that failed open would leave an application believing its stream is protected while it keeps flowing — which is worse than having no guard at all, because nobody goes looking.

Know the blast radius before you write one. The aggregator is a process-wide singleton over every discovered guard type, so a single guard that cannot be constructed at all — an open generic, or one whose constructor takes something the subscription scope cannot resolve — denies every observable query in the application, on every transport, for as long as the process runs. The only symptom is the Error log; clients simply see their subscriptions end as unauthorized. Cover a new guard with a spec that exercises its real constructor.

One exception is carved out: if the subscription’s own CancellationToken is canceled while the guard is running, the resulting OperationCanceledException is the client going away, not a guard failing. The emission is still withheld and the subscription still ends, but nothing is logged as a failure — otherwise every closed tab would produce an authorization error.

Cost

The guard runs on every emission of every subscription. Keep it fast: prefer cached state, a local revocation list, or a short-TTL lookup over a network round trip per emission.

The guard instance itself is constructed per emission. Guards are resolved with GetServiceOrCreateInstance, and Arc’s IFoo → Foo convention does not match a guard named for what it decides (SessionMustStillBeActive implements IGuardObservableQueryEmission, not ISessionMustStillBeActive), so unless the application registers the type explicitly it is newly constructed — constructor and all — for every emission, and never disposed. Keep the constructor trivial, and put any cache in an injected singleton rather than in a field on the guard, where it would be thrown away and rebuilt each time.

Resolve collaborators through the constructor. They come from the subscription’s own scope and are disposed with it, so a guard can safely hold a scoped session store or tenant-aware service.

Where it does not help

Guards protect emitted values, including observable HTTP snapshots and non-streaming hub results; they do not replace subscription-time authorization or safe data selection. Direct HTTP non-observable queries do not invoke emission guards: those requests authenticate again on each call. Apply query authorization to every query and keep read-model interception’s opt-out boundaries in mind.

On a WebSocket the principal is frozen at the handshake — the protocol offers no way to present fresh credentials on an established connection. The identity a guard receives on a WebSocket subscription is the one captured when the socket was upgraded, and it does not change for the life of that connection.

That does not make the guard useless there; it makes it a lookup rather than a re-read. Take the identity from the context and ask your own source of truth — a session store, a token introspection endpoint, a revocation list — whether that identity is still good.