MeshWeaver models data mutations as reactive messages flowing through a typed workspace. Rather than imperative method calls, you describe what changed and the workspace propagates the delta to every subscriber automatically — no polling, no manual refresh.

Scope: this page covers typed-entity CRUD — instances (TodoItem, Project, …) living in a workspace's EntityStore collections. It is not the API for mutating MeshNodes. A MeshNode's lifecycle goes through CreateNodeRequest/DeleteNodeRequest/MoveNodeRequest, and a MeshNode's content is mutated exclusively via workspace.GetMeshNodeStream(path).Update(current => modified).Subscribe(...) — see Node Operations and CQRS — Queries vs. Content Access.

CRUD pipeline: every mutation passes through Access Control and Validation before reaching the Workspace, which persists the change and fans it out to all subscribers in real time.

Data Model

Every workspace maintains an EntityStore: a map of named collections, each holding a set of typed entity instances keyed by their ID.

EntityStore
├── Collections["TodoItems"] = InstanceCollection
│   ├── Instances["todo-1"] = TodoItem { Id: "todo-1", Title: "Task 1" }
│   └── Instances["todo-2"] = TodoItem { Id: "todo-2", Title: "Task 2" }
└── Collections["Projects"] = InstanceCollection
    └── Instances["proj-1"] = Project { Id: "proj-1", Name: "Alpha" }

An InstanceCollection is a container for instances of a specific entity type, mapping IDs to objects.

Create

DataChangeRequest

The primary way to add new entities is to send a DataChangeRequest with the Creations payload:

var newTodo = new TodoItem
{
    Id = Guid.NewGuid().ToString(),
    Title = "Learn MeshWeaver",
    Status = TodoStatus.Pending
};

var request = DataChangeRequest.Create([newTodo], changedBy: "user-123");
workspace.RequestChange(request, delivery);

Create Flow

Read

Everything is a live stream. The same primitive serves both "keep me updated" and "give me the current state" — there is no separate snapshot API to keep consistent.

Canonical node access — GetMeshNodeStream(path)

Any node — own, local, or remote — is read through workspace.GetMeshNodeStream(path). It is live and authoritative (served by the owning hub, no index lag), and the same handle accepts writes via .Update(...):

// Live subscription — emits the current node, then every subsequent change:
workspace.GetMeshNodeStream("acme/TodoLists/work")
    .Where(node => node is not null)
    .Subscribe(node =>
    {
        var list = node.ContentAs<TodoList>(hub.JsonSerializerOptions, logger);
        // render / react
    });

// One-shot read — same primitive, complete after the first emission:
workspace.GetMeshNodeStream(path)
    .Where(node => node is not null)
    .Take(1)
    .Timeout(TimeSpan.FromSeconds(10))
    .Subscribe(node => /* use the snapshot */);

Never read a known node through QueryAsync — queries are an eventually-consistent index for sets, stale right after a write. See CQRS — Queries vs. Content Access.

Reactive Subscription — typed collections

Subscribe to a typed collection as an IObservable that emits the full current set on every change:

workspace.GetObservable<TodoItem>()
    .Subscribe(todos =>
    {
        Console.WriteLine($"Todos updated: {todos.Count}");
        foreach (var todo in todos)
            Console.WriteLine($"  - {todo.Title}");
    });

For a one-shot collection snapshot, compose on the same stream — workspace.GetObservable<TodoItem>().Take(1).Timeout(...) — instead of a request/response round-trip.

Reference Types

Choose the right reference type for your query pattern:

Unified Reference Paths

Path-based references give a uniform addressing scheme across entity and content resources. The prefix is separated by a colon, and what follows it starts with the owning address — {prefix}:{addressType}/{addressId}/…. A path with no recognised colon prefix falls through to area, so a slash-separated "data/TodoItems" is parsed as an area reference, not data.

var allTodosRef = "data:app/my-app/TodoItems";           // entire collection
var todoRef     = "data:app/my-app/TodoItems/todo-1";    // specific entity
var fileRef     = "content:app/my-app/uploads/doc.pdf";  // file content
var areaRef     = "area:app/my-app/Dashboard";           // layout area

The three recognised prefixes are data:, area: and content: (ParseUnifiedPath). There is no schema: prefix.

Virtual Paths

Define computed data sources that combine or transform real collections:

.WithVirtualPath("TodoSummary", (workspace, entityId) =>
{
    var todos = workspace.GetStream(typeof(TodoItem));
    var users = workspace.GetStream(typeof(User));

    return Observable.CombineLatest(todos, users, (t, u) =>
    {
        // Compute summary by joining data
        return new TodoSummary { ... };
    });
})

Virtual paths participate in real-time propagation just like ordinary collections.

Update

DataChangeRequest

Pass updated entity instances in the Updates payload. By default, changes are merged into the existing record:

var updatedTodo = existingTodo with
{
    Title = "Updated Title",
    Status = TodoStatus.InProgress
};

var request = DataChangeRequest.Update([updatedTodo], changedBy: "user-123");
workspace.RequestChange(request, delivery);

An instance with no key value is skipped, not fatal. When a batch carries an instance whose key resolves to null, that instance is reported in the activity log under activity.dataUpdate.nullKeyInstancesSkipped (with its count and collection) and skipped, while every keyed instance of the same batch still applies and the data stream stays live for later writes. A type with no key MAPPING at all is a configuration error and still fails the change.

Update Options

Control whether a change merges or replaces the entire collection:

// Merge (default) — only the supplied instances change
var mergeRequest = DataChangeRequest.Update(
    updates: [updatedTodo],
    changedBy: "user-123",
    options: new UpdateOptions { Snapshot = false }
);

// Snapshot — the entire collection is replaced by the supplied list
var snapshotRequest = DataChangeRequest.Update(
    updates: allTodos,
    options: new UpdateOptions { Snapshot = true }
);

Updating node content — GetMeshNodeStream(path).Update(...)

When the thing you are updating is a node's content (not an entity inside a collection), use the canonical mutation API — the same handle you read from:

workspace.GetMeshNodeStream("acme/TodoLists/work").Update(node =>
{
    var list = node.ContentAs<TodoList>(hub.JsonSerializerOptions, logger);
    if (node.Content is not null && list is null) return node;  // never clobber unreadable content
    list ??= new TodoList();
    return node with { Content = list with { Title = "Updated Title" } };
})
.Subscribe(_ => { }, ex => logger.LogWarning(ex, "update failed"));

Update is cold — the write only happens on Subscribe. Cross-hub writes ship an RFC 7396 JSON-merge patch to the owning hub, so concurrent writers touching different fields both land.

Workspace Extension Methods

Convenience wrappers for the most common patterns:

// Single entity
workspace.Update(updatedTodo, delivery);

// Multiple entities
workspace.Update([todo1, todo2, todo3], delivery);

Delete

DataChangeRequest

Pass the full entity object (not just the ID) so the workspace can resolve and remove the correct instance:

var request = DataChangeRequest.Delete([todoToDelete], changedBy: "user-123");
workspace.RequestChange(request, delivery);

Workspace Extension Methods

workspace.Delete(todoToDelete, delivery);
workspace.Delete([todo1, todo2], delivery);

Deleting a node (as opposed to an entity in a collection) is a lifecycle operation: hub.Observe<DeleteNodeResponse>(new DeleteNodeRequest(path)).Subscribe(...) — see Node Operations.

Delete only if unchanged — ExpectedVersion

A caller that reads a node, decides from what it read that the node may go, and then deletes it has a race: another writer can change the node between that read and the delete, and the delete removes the change. You cannot close that race from the caller's side. Checking again just before deleting leaves the same window. So name the version you read, and let the owner check it:

var request = new DeleteNodeRequest(path) { ExpectedVersion = node.Version };
hub.Observe(request, o => o.WithTarget(hub.NodeOperationTarget()))
    .Subscribe(
        d =>
        {
            if (d.Message.RejectionReason == NodeDeletionRejectionReason.Conflict)
            {
                // It changed after we read it. Nothing was deleted: re-read and decide again.
            }
        },
        ex => logger.LogWarning(ex, "Delete of {Path} did not answer", path));

• Where it is checked: by the owning node hub, in its commit turn. The delete first seals the path against the owner's own writes and opens the subtree-deletion scope, then asks the owning hub to compare ExpectedVersion with the node's live version. The owner answers on the same turn every own write and every cross-hub patch commits on. So every write the owner has already acknowledged is in the version it compares, and any later write is refused instead of being acknowledged: an own write is refused in its turn (SubtreeDeletionInFlightException), and a cross-hub patch is refused at its durable flush (NotFound, "Node being deleted"). The check and the delete are one step: nothing acknowledged can land between them.
• Why the live version, not the stored one. An own write is acknowledged from the owner's in-memory commit, about one persistence-sampler interval before it is stored. A check against the stored row matched the old version in that window, and the delete removed a change its writer had been told was accepted (#6673).
• A mismatch answers Success = false with RejectionReason = Conflict, the same optimistic-concurrency fact as MeshNodeErrorCode.Conflict on an update. Nothing is deleted.
• An absent node still answers the idempotent success AlreadyAbsent (Deleting What Is Already Gone). The postcondition holds, whatever version the node last had.
• And checked again by the store, in the delete itself. The root row is removed with a compare-and-delete (IStorageAdapter.DeleteIfVersionAtMost; on Postgres DELETE … WHERE version <= $expected and the row count). A row newer than the confirmed version was written after the owner's check, by a writer the seal did not stop (another process, or anything that writes the store without passing the owner). The store refuses to remove it, and the delete answers Conflict naming the stored version. A row at or below the confirmed version is the node the deleter read, one sampler flush behind, so it is removed. The guarantee therefore no longer depends on the seal alone, which is held only in the deleting process. Only a row in writable storage counts as a survivor. Shipped content that a read-only provider serves under a removed writable row is not this delete's to remove, so it is no conflict.
• When the same path has copies in several writable providers, a copy newer than the confirmed version refuses the whole delete and nothing is removed. The version the conflict names is the first copy a read finds, which can be the older one. Removals across separate stores are not atomic: a newer copy written into a second provider while the delete runs is refused there, but an older copy already removed is not restored.
• Root only. For a recursive delete, only the node at Path is checked. The cascade's per-descendant requests never carry the value. The cascade removes children before their parent, so when the store refuses the root, the descendants are already gone. The response names them, and the root survives.
• Which version to name. The version you read, from the node stream or from an acknowledged write. It is compared with what the owner holds, so a version that is acknowledged but not yet stored matches. The node a cross-hub GetMeshNodeStream(path).Update(...) emits carries the version the owner committed (PatchDataResponse.CommittedVersion), so it can be named directly. Its content is still your writer's view: fields that a concurrent writer merged in on the owner are not in it. An owner on an image older than this field does not send it, and the emitted node then carries your base version, one behind the owner. Naming that version is refused as Conflict; re-read the node stream and use its version.
• No verdict is a refusal. If the owner cannot be asked and the node is still stored, the delete answers Unavailable and removes nothing. It never treats an unanswered check as a match. The same holds when the storage read that would tell "gone" from "unreachable" fails as well.
• A match must be confirmed. The owner's "matches" answer echoes the version it compared (ValidateDeleteResponse.ConfirmedVersion). During a rolling update, an owner on an older image does not know ExpectedVersion: it ignores the field, runs the ordinary delete validators and answers a plain "valid". That answer carries no echo, so the delete reads it as Unavailable, not as a match.
• A recycling owner is asked again. If the owner's activation is torn down while it compares, it answers ShuttingDown, as it does for an ordinary delete check. The delete then asks the next activation, within the same time budget.
• Not covered yet (tracked on #6856): the seal is still held only in the deleting process. When the owning activation runs on another silo, that silo can acknowledge an own write after the comparison. The store refuses the delete only if that write was flushed before it. If the delete reaches the store first, the later flush stores the node again: nothing acknowledged is lost, but the delete has reported success for a node that exists again. Only the in-memory and Postgres stores implement DeleteIfVersionAtMost atomically (Postgres in MeshWeaver.Plugins, as one conditional DELETE whose row count is the verdict; its path router forwards the call). The file-system store (and the caching adapter over it) and Snowflake use the interface default: they read, compare, then delete, which checks the version but not atomically, so a write landing between the read and the delete is removed. On the file-system backend that matters only with more than one writer per directory; it is a single-writer store. And when the store refuses a delete because of a write from another process, the owner here adopts that write only when its change notification, or a declared feed gap, arrives. Until then the node stream can still show the older version, and a retry naming it is refused again. That is safe, because nothing is removed, and the conflict names the stored version.
• Omitted (null, the default), the delete removes whatever version is there, exactly as before.

Data Validation

Attach validators to enforce business rules before any change is applied. The workspace calls every registered validator and returns DataValidationResult.Invalid(...) to the caller if any rule is violated.

public class TodoValidator : IDataValidator
{
    // Reactive, never Task<T> — this runs on the hub.
    public IReadOnlyCollection<DataOperation> SupportedOperations { get; } =
        [DataOperation.Create, DataOperation.Update];

    public IObservable<DataValidationResult> Validate(DataValidationContext context)
    {
        if (context.Entity is TodoItem todo && string.IsNullOrEmpty(todo.Title))
            return Observable.Return(
                DataValidationResult.Invalid("Title is required"));

        return Observable.Return(DataValidationResult.Valid());
    }
}

Register validators in DI — the workspace resolves every registered IDataValidator:

services.AddScoped<IDataValidator, TodoValidator>();

Access Control

Restrict operations based on user context. Access restrictions run before validation, so unauthorized requests are rejected early.

Global restriction — apply a rule to all operations in a data source:

.AddData(data => data
    .WithAccessRestriction(
        (action, context, accessCtx) =>
        {
            if (action == AccessAction.Read)
                return Observable.Return(true);          // anyone can read

            return Observable.Return(accessCtx.UserContext != null); // writes require login
        },
        "RequireAuthForWrites"
    )
)

Type-specific restriction — limit access at the entity level:

.AddSource(src => src
    .WithType<TodoItem>(type => type
        .WithAccessRestriction((action, ctx, accessCtx) =>
        {
            var todo = ctx as TodoItem;
            // Only the owner may modify their own todos
            return Observable.Return(
                todo?.OwnerId == accessCtx.UserContext?.ObjectId
            );
        }, "OwnerOnly")
    )
)

Configuration Example

A complete data source setup showing multiple types, a virtual path, a validator, and an access rule:

.AddData(data => data
    .AddSource(src => src
        .WithType<TodoItem>(type => type
            .WithKey(todo => todo.Id)
            // Every WithInitialData overload takes an IObservable factory (or a fixed
            // collection) — there is no Task/async overload. A real I/O leaf goes
            // through an IIoPool, never Observable.FromAsync.
            .WithInitialData(() => dbPool.Invoke(ct => LoadTodosFromDatabaseAsync(ct)))
        )
        .WithType<Project>(type => type
            .WithKey(proj => proj.Id)
        )
    )
    .WithVirtualPath("Dashboard", ComputeDashboard)
    .WithAccessRestriction(RequireAuthentication, "Auth")
)

// Validators are registered in DI, not on the data configuration:
services.AddScoped<IDataValidator, TodoValidator>();

Real-Time Synchronization

Every CRUD operation automatically propagates to all current subscribers. Clients never need to re-query; the workspace pushes the updated collection as soon as the change is applied.

Quick Reference

Message Types

Best Practices

1. Use typed observables — prefer GetObservable<T>() over raw streams for compile-time safety.
2. Check the response — inspect DataChangeResponse.Status (Committed / Failed) before assuming success; the detail is in .Log. Note a Warning status still COMMITS.
3. Batch related changes — group inserts, updates, and deletes into a single DataChangeRequest for atomic delivery.
4. Register validators — enforce data integrity at the data layer rather than in each call site.
5. Protect with access restrictions — declare who may read or write each type alongside the type configuration.
6. Prefer subscriptions over polling — reactive streams keep UIs in sync with zero manual refresh logic.

Live Demo

The cell below builds a small in-memory summary table using the reference types described above, rendered directly in this page:

var rows = new[]
{
    new { Type = "EntityReference",      Purpose = "Single entity by ID",         Example = "new EntityReference(\"TodoItems\", \"todo-1\")" },
    new { Type = "CollectionReference",  Purpose = "All entities in collection",   Example = "new CollectionReference(\"TodoItems\")" },
    new { Type = "CollectionsReference", Purpose = "Multiple collections at once", Example = "new CollectionsReference(\"TodoItems\", \"Projects\")" },
};

MeshWeaver.Layout.Controls.DataGrid(rows)
    .WithColumn(new MeshWeaver.Layout.DataGrid.PropertyColumnControl<string>
        { Property = "type" }.WithTitle("Reference Type"))
    .WithColumn(new MeshWeaver.Layout.DataGrid.PropertyColumnControl<string>
        { Property = "purpose" }.WithTitle("Purpose"))
    .WithColumn(new MeshWeaver.Layout.DataGrid.PropertyColumnControl<string>
        { Property = "example" }.WithTitle("Example"))

See Also

• Query Syntax — Search and filter nodes
• Unified Path — Path-based data addressing
• Data Binding — Connect UI controls to data
• Editor Control — Generate forms from records