Observable query hub protocol

Endpoints

The demultiplexer carries multiple named observable queries over one connection. Both WebSocket and SSE are multiplexed. Arc maps four fixed routes:

Method/transport	Route	Purpose
WebSocket upgrade	/.cratis/queries/ws	Bidirectional subscription and result messages
GET, SSE	/.cratis/queries/sse	Long-lived output stream
POST, JSON	/.cratis/queries/sse/subscribe	Add/replace a subscription on an SSE connection
POST, JSON	/.cratis/queries/sse/unsubscribe	End a subscription on an SSE connection

A direct per-query SSE GET is a different protocol: it streams plain QueryResult frames for that route, not hub envelopes. See cURL streaming.

Message types

ObservableQueryHubMessage.Type has its own JsonStringEnumConverter. The server emits the following PascalCase string values, not the camelCase field names or numeric enum constants. Numeric values are listed only to identify the enum; use strings in new protocol clients.

type	Enum value	Purpose/payload
Subscribe	0	Client sends a subscription request in payload
Unsubscribe	1	Client ends queryId
QueryResult	2	Server sends a QueryResult in payload
Unauthorized	3	Server denies the subscription
Error	4	Server sends an error string in payload
Ping	5	Keep-alive; timestamp is Unix milliseconds
Pong	6	Echo of a ping timestamp
Connected	7	Connection ID in payload, plus capability/keep-alive metadata

Envelope properties are type, queryId, revision, payload, timestamp, keepAliveIntervalMs, and supportsSubscriptionRevisions. Not every field is meaningful on every message; nullable/default fields may be omitted according to serializer configuration.

WebSocket transport

After upgrade, the server sends Connected before reading subscriptions. Illustrative frame (null fields omitted for readability):

{
  "type": "Connected",
  "payload": "ws-1",
  "supportsSubscriptionRevisions": true,
  "keepAliveIntervalMs": 30000
}

For the Banking.Accounts.DebitAccount.ObserveAccounts method in the model-bound example, send:

{
  "type": "Subscribe",
  "queryId": "accounts-list",
  "revision": 1,
  "payload": {
    "queryName": "Banking.Accounts.DebitAccount.ObserveAccounts",
    "arguments": { "minimumBalance": "0" },
    "page": 0,
    "pageSize": 25,
    "sortBy": "name",
    "sortDirection": "asc",
    "transferMode": "delta"
  }
}

queryName is the fully qualified performer name, not its HTTP path. queryId is chosen by the client and is unique per active subscription within that connection. Arguments are a string-value dictionary; nested JSON/array binding is not added by this protocol. Paging, sorting, and transfer mode are optional request properties.

Updates carry type: "QueryResult", the matching queryId and revision, and the result envelope in payload. Subscribe again with a different queryId to observe another query on the same connection. To end the first subscription:

{ "type": "Unsubscribe", "queryId": "accounts-list", "revision": 1 }

A client Ping receives a Pong with its timestamp. Connection shutdown disposes the connection’s subscriptions; it is not a durable replay checkpoint.

SSE transport

1. Open GET /.cratis/queries/sse and keep it open.
2. Read the first Connected frame and retain its GUID connection ID.
3. POST a subscription body to /.cratis/queries/sse/subscribe using that ID.
4. Read updates on the original GET, correlated by queryId and revision.
5. POST unsubscribe when done; close the GET to end all its subscriptions.

Illustrative subscribe body; replace the connection ID with the one just received:

{
  "connectionId": "11111111-1111-1111-1111-111111111111",
  "queryId": "accounts-list",
  "revision": 1,
  "request": {
    "queryName": "Banking.Accounts.DebitAccount.ObserveAccounts",
    "arguments": { "minimumBalance": "0" },
    "transferMode": "full"
  }
}

The unsubscribe body is:

{
  "connectionId": "11111111-1111-1111-1111-111111111111",
  "queryId": "accounts-list",
  "revision": 1
}

Send both POSTs as application/json. The validated control requests return 400 for missing required values/invalid revisions, 404 for unknown connections, and normally 200 after processing. Subscribe returns 401 when authorization denies the query; denial is also delivered as an Unauthorized frame. A 200 control response is not proof that a first data result arrived—observe the stream’s results/errors.

?query=... on the SSE GET does not subscribe. On reconnect, wait for the new Connected ID and re-create desired subscriptions. Connection state is server-process-local; route the GET and its control POSTs to the same instance in a multi-instance deployment.

Revisions and compatibility

When Connected.supportsSubscriptionRevisions is true, use positive monotonically increasing revisions for each queryId. A higher subscribe revision replaces older work; duplicate or stale subscribes are ignored. Unsubscribe uses the exact revision being canceled, and can arrive before a delayed subscribe to tombstone it. An older unsubscribe cannot tear down a newer subscription. Discard stale result/error/denial frames on the client.

Legacy clients may omit revisions. Once a query ID has become revision-aware, revisionless operations cannot replace/cancel that state. Older servers may omit Connected on WebSocket or omit the capability field; clients requiring compatibility must support the legacy behavior rather than indefinitely waiting for an advertisement those servers never send.

Tombstones are bounded: the current implementation retains them for two minutes and at most 1,024 inactive entries per connection. This is an ordering window, not indefinite deduplication or durable resumption.

Transfer modes

For subject-backed collections, omitted transferMode uses legacy snapshot-plus-delta behavior; full sends snapshots only; delta sends an initial snapshot then changes without full data. See change streams for the first/subsequent emission matrix. Arbitrary subjects do not automatically implement paging just because subscription metadata requests it.

Paging metadata belongs to collection results. Single-result subjects, including strings and value types, report PagingInfo.NotPaged rather than copying TotalItems from an inner observation. Collection delta emissions keep their paging metadata even when they omit the full data snapshot. This only controls the result envelope; it does not normalize incoming paging arguments or change how an application reads its sources. For MongoDB queries that compose observations, see auxiliary sources.

Restricting WebSocket origins

By default the hub, and every other Arc observable query WebSocket route, accepts connections from any origin. To restrict them, set Cratis:Arc:Query:AllowedWebSocketOrigins:

{
    "Cratis": {
        "Arc": {
            "Query": {
                "AllowedWebSocketOrigins": [ "https://portal.example.com" ]
            }
        }
    }
}

Once the setting is present (even as an empty list set in code), a connection is accepted when it has no Origin header, when its Origin is the origin the request was made to (same scheme, host and effective port), or when its Origin is listed. Anything else, including an empty, null or duplicated Origin, gets 403 before the connection is accepted and before any query runs. A * entry accepts every origin. Loopback origins get no special treatment, so a development proxy on another port must be listed.

The request’s own origin is taken from the Host header and the connection scheme as ASP.NET Core sees them. Behind a reverse proxy that rewrites Host or terminates TLS, apply the forwarded headers with ASP.NET Core’s UseForwardedHeaders, or list the public origin.

Browsers cannot send an Authorization header on a WebSocket, so applications that authenticate browsers with cookies should configure this setting. Arc logs a startup warning when cookie authentication is configured and the list is empty.

Authorization and exposure

The hub transport endpoints themselves allow anonymous access so one connection can carry both public and protected queries. Before running each query, the hub applies the host’s default authorization: ASP.NET Core’s FallbackPolicy, or the Arc Core host’s requirement for an authenticated caller when authentication handlers are registered. A query marked with Arc’s [AllowAnonymous], or whose effective declaration consists entirely of opted-in guest policies, bypasses that host fallback just as its direct endpoint does. The query pipeline still evaluates its own authorization filters and guest policies. A fallback denial or failure sends Unauthorized without running the query; SSE subscribe also returns 401. The SSE GET and unsubscribe routes do not execute queries.

The same model-bound authorization rules apply to a subscription as to a one-shot request. WebSocket credentials come from the live upgrade request; SSE credentials come from the live subscribe POST. Consistently send the application’s normal credentials on connection/control requests.

The SSE control handlers verify that the POST caller is the caller that opened the GET stream. The identity is captured once when the stream opens - while that request is still in flight, so the retained context is never read again - and each subscribe and unsubscribe is compared against it on the identity identifier claim and authentication state. A mismatch returns 404, exactly as an unknown connection ID does, so the endpoints never confirm that another caller’s connection exists.

The display name is not part of that comparison, because it is mutable and a stream held open across a token refresh would otherwise stop accepting its own caller’s control requests. The tenant is not part of it either: a control POST is authorized with the tenant resolved from that POST, not the one the connection was opened with.

A connection opened by an unauthenticated caller carries no identity to bind to, so ownership is not distinguishable between anonymous callers. Treat the SSE connection ID as sensitive connection-control data rather than as an authorization policy, and do not publish or unnecessarily log connection IDs. Query authorization still checks the subscribe POST’s right to read the query, which is a separate question from ownership of the destination stream. The health feed includes connection/subscriber metadata; it allows anonymous access in Development by default and requires authentication elsewhere, with configurable authentication and roles.

The initial verdict does not automatically revoke a long-lived stream. Use emission guards to check current permission/session state during delivery. Guards also run before the hub sends a non-streaming query’s one-shot result, with IsFirstEmission=true and IsSnapshot=true. A denied snapshot sends Unauthorized; a suppressed snapshot sends no result, error, or denial frame and ends without a continuing subscription. An SSE subscribe POST still returns 200 for suppression.

Keep-alive

ArcOptions.Query.KeepAliveInterval defaults to 30 seconds. Idle WebSocket/SSE connections receive a Ping; ongoing data suppresses unnecessary keep-alives. Zero or negative disables keep-alive, advertised as keepAliveIntervalMs: 0 on Connected. Clients should derive their idle threshold from that advertisement rather than hard-coding 30 seconds.

See frontend multiplexing for the supplied client rather than implementing the lifecycle from scratch.