Model-bound observable queries

A page that should stay current needs a producer of updates, not repeated manual refreshes. An observable query exposes that producer through Arc. The client chooses a snapshot, direct SSE/WebSocket, or a multiplexed hub subscription; a return type alone does not establish a connection.

Basic observable query

This alternative declaration uses the shared AccountId and AccountName concepts, an Arc host configured with Cratis.Arc.MongoDB, a registered collection, and MongoDB change-stream support (a replica set or sharded deployment). Observe() lives in the MongoDB.Driver namespace. This is ordinary database observation in standalone Arc, not a Chronicle projection.

using System.Collections.Generic;
using System.Reactive.Subjects;
using Cratis.Arc.Authorization;
using Cratis.Arc.Queries.ModelBound;
using MongoDB.Driver;

namespace Banking.Accounts;

[ReadModel]
[Roles("AccountReader")]
public record DebitAccount(AccountId Id, AccountName Name, decimal Balance)
{
    [Path("/api/accounts/observe")]
    public static ISubject<IEnumerable<DebitAccount>> ObserveAccounts(
        IMongoCollection<DebitAccount> collection,
        decimal minimumBalance = 0) =>
        collection.Observe(account => account.Balance >= minimumBalance);

    [Path("/api/accounts/observe-single")]
    public static ISubject<DebitAccount> ObserveAccount(
        AccountId id,
        IMongoCollection<DebitAccount> collection) =>
        collection.ObserveSingle(account => account.Id == id);
}

When There Is No Matching Document

ObserveSingle() and ObserveById() emit null — not an error, and not a completed observable — when there is no document to report: the document was deleted, an update moved it out of the filter, or the initial query never found one. The subscription stays open, so if a document with the same key reappears later, subscribers start receiving it again.

This is also what happens when a [ReadModel] marked [RemovedWith<T>] is removed: removal hard-deletes the backing document, and an active subscriber sees that removal as this same null emission — “the read model was removed” is not a special case to handle separately.

Guard against it on the frontend the same way you guard against “not loaded yet” — with result.hasData (or result.isReady if you need to tell “no result yet” apart from “ready, but nothing matches”):

const [result] = GetAccountObservable.use(accountId);

if (!result.isReady) {
    return <Spinner />;
}

if (!result.hasData) {
    return <NotFound />;
}

return <AccountDetails account={result.data} />;

See Observing Collections for the MongoDB-level detail.

The method must still return the declaring read model or a supported wrapper around it. Task<ISubject<T>> is allowed; arbitrary IObservable<T> is not a model-bound discovery shape. See return types.

Custom observable logic

Return the provider subject directly when it expresses the read. Rx operators such as Select, CombineLatest, or Sample produce observables; adapting them to Arc’s subject contract must preserve both subscription lifetime and terminal errors. This keeps derived streams safe to disconnect, not just able to forward values.

Subscription lifetime

Direct transports dispose the subscription they receive; the hub also manages per-subscription resources. A composition must therefore honor two contracts:

Subscription-owned teardown: the disposable returned by Subscribe must release every upstream subscription. Keep the disposable from .Subscribe(replaySubject); implementing IDisposable on a wrapper alone does not connect it to downstream teardown.
Terminal error propagation: deliver errors to downstream subscribers and release upstream resources. Direct SSE and WebSocket transports can call the returned subject’s OnError after interception or delivery failures. An onNext-only adapter uses a default throwing error handler; explicitly preserve the terminal-error path rather than swallowing it.

Test interception failures through both direct transports, terminal error delivery, and repeated subscribe/disconnect cycles. Verify watcher counts return to baseline.

Paging and failures

MongoDB Observe() reads the Arc query context and performs provider-specific paging. Arbitrary subjects do not get automatic per-emission slicing.

For the collection Observe() and ObserveSingle() producers above, MongoDB watcher failures are logged and followed by completion/disposal, not forwarded through OnError. Per-change processing failures are logged separately and the watcher may continue. Therefore Do(onError: ...) cannot guarantee notification of these source failures. Downstream Rx operator errors are a separate error channel. Monitor provider logs and stream health; completion does not prove that the data is current.

Waiting for the first HTTP result

For MongoDB snapshots, use waitForFirstResult=true so Arc subscribes, receives the first emission, and disposes that subscription. Waiting defaults to 30 seconds; a positive waitForFirstResultTimeout overrides it in seconds, and timeout returns 408.

A no-wait snapshot neither subscribes nor disposes the subject. MongoDB’s LifetimeAwareSubject has no readable Value property, even after buffering an emission, so this request returns 202 with isReady: false. Because Observe() starts its watcher eagerly, it can leave that watcher running. Reserve no-wait snapshots for producers with a readable current value and verified cleanup on the no-subscription path.

curl --include 'https://localhost:5001/api/accounts/observe?waitForFirstResult=true&waitForFirstResultTimeout=10'

Use a trusted local certificate and your application’s normal authentication for this protected example. This request waits for a first emission, not a change since the previous request. See cURL workflows for SSE and bounded repeated polling.

Authorization and frontend integration

Authorization runs when subscribing. Use emission guards when permissions must be rechecked during delivery. Generated proxies and React observable hooks manage client subscriptions; reconnecting creates a new subscription, not a durable event-log replay.