Subscriptions

Subscriptions deliver a real-time stream of change notifications for all entities of a given kind. They are the primary way to observe state changes in ConvergeDB.

Subscribing

C#

await foreach (var change in players.SubscribeAsync(bootstrap: true))
{
    Console.WriteLine($"{change.Type}: {change.Entity?.Name} (v{change.Version})");
}

Rust

// Coming soon

The SubscribeAsync method returns an IAsyncEnumerable<EntityChange<T>> that yields notifications as they arrive.

Parameters

Parameter	Default	Description
bootstrap	false	If true, the server sends a snapshot of all existing entities before live notifications. See Bootstrap.
includePrevious	false	If true, Updated notifications include the previous entity state. See Change Tracking.
reconnectMode	LiveOnly	Controls what happens on reconnection. See Reconnection.
ct	default	Cancellation token. Cancelling unsubscribes and stops the stream.

Notification types

Each notification has a Type indicating what happened:

Type	Meaning
Created	A new entity was created (or re-created from a tombstone).
Updated	An existing entity’s field data changed.
Deleted	An entity was tombstoned (all sources retracted).
Bootstrap	An entity snapshot from the bootstrap scan. See Bootstrap.
BootstrapComplete	Sentinel: the bootstrap snapshot is fully delivered. Entity is null. See Bootstrap.

Notification contents

Every notification carries:

Property	Description
Type	The notification type (see above).
Entity	The current entity state (typed as T). Null for Deleted.
EntityId	The 32-byte entity ID.
Version	The entity’s current version (monotonically increasing).
SourceSet	Bitmask of which sources assert this entity.
ChangedFields	Bitmask indicating which fields changed. See Change Tracking.
PreviousEntity	The previous entity state, if includePrevious was true and the type is Updated.
Metadata	Opaque metadata from the write that triggered this notification. See Metadata.

Filtering

In v1, subscriptions filter by entity kind only. You subscribe to all entities of a given kind and receive all changes. Field-level filtering (subscribing only to entities where a specific field matches a predicate) is planned for v2.

If you need to filter on the client side, check the notification’s fields in your processing loop:

C#

await foreach (var change in players.SubscribeAsync(bootstrap: false))
{
    if (change.Entity?.Health > 0)
    {
        // Process only living players
    }
}

Rust

// Coming soon

Backpressure and slow subscribers

Each subscription has a bounded notification buffer (default: 4,096 notifications). Backpressure behavior differs between live notifications and bootstrap:

Live notifications: If the subscriber falls behind and the buffer fills, the subscriber is disconnected immediately by the server. On reconnection, the subscriber must re-subscribe. If ReconnectBootstrapMode.Full is configured, the subscriber automatically re-bootstraps to recover missed state. See Reconnection.

During bootstrap: The server applies backpressure rather than disconnecting immediately. When the buffer is full, the server retries delivery with brief pauses, giving the subscriber time to drain. If the subscriber remains stalled for 5 seconds (e.g. a dead TCP connection), it is disconnected. This means the buffer size does not limit how many entities can be bootstrapped — kinds with millions of entities bootstrap correctly regardless of the buffer size.

To avoid disconnection during live notifications:

• Process notifications promptly in the await foreach loop.
• Offload expensive work to a background queue rather than processing synchronously in the notification loop.
• Increase SubscriberBufferSize in ConvergenceOptions if your workload produces large notification bursts.

Error handling

Wrap individual notification processing in a try-catch so that a single failure does not kill the subscription. This is especially important for services that write to external databases or call external APIs:

C#

await foreach (var change in players.SubscribeAsync(bootstrap: true, ct: ct))
{
    try
    {
        await ProcessChangeAsync(change, ct);
    }
    catch (OperationCanceledException) { throw; } // Propagate cancellation.
    catch (Exception ex)
    {
        logger.LogError(ex, "Error processing {Type} for entity", change.Type);
        // Continue processing next notification.
    }
}

Rust

// Coming soon

Re-throw OperationCanceledException so that cancellation still stops the subscription cleanly.

Subscription lifecycle

The subscription is active for the lifetime of the await foreach loop. When the loop exits (via cancellation, break, or exception), the SDK sends an UNSUBSCRIBE frame to the server and releases the subscription resources.

You can also manage subscriptions manually via SubscriptionHandle. See Building Correct State for advanced patterns.

Per-operation events

Subscriptions deliver coalesced state: multiple writes within a coalescing window are merged into a single notification. If you need to observe every individual ASSERT, PATCH, or RETRACT operation with its source identity and a total ordering, see Event Streams.