Initialization Gates

TL;DR — A hub may need to hold back inbound traffic until its own data is loaded. Declare a gate at config time with WithInitializationGate(name, letThroughPredicate), then open it on the first emission of a reactive observable — either inside .Select(...) or in a Subscribe callback. Never bridge to await Task<T> to open a gate; that captures the calling scheduler and deadlocks the hub action block. The condition must be expressible as a non-blocking observable emission.

A gate is startup-only: it defers inbound deliveries and opens once. It cannot suppress anything the hub does in steady state, and gating on work that WRITES nodes can defer the very response that opens it. See What a gate is NOT for before reaching for one.

What is a gate?

A hub initialization gate is a delivery-pipeline filter installed at startup. While the gate is closed, every inbound delivery is queued — except those that pass the gate's bypass predicate. When hub.OpenGate(name) is called, the queue drains in order. Closing again after opening is a no-op: gates are one-way.

Register a gate on MessageHubConfiguration. The signature is WithInitializationGate(string name, Predicate<IMessageDelivery>? allowDuringInit = null):

config.WithInitializationGate(
    MeshNodeExtensions.MeshNodeInitGateName,
    d => d.Message is CreateNodeRequest);

hub.OpenGate(name) is idempotent: call it from multiple branches (success, failure, no-persisted-node paths) without worry.

When do you need a gate?

You need a gate when all three of these are true:

1. The hub's stream or reducer sources its initial value from persistence or another non-trivial load.
2. The hub also accepts request traffic that queries that stream.
3. Requests can arrive before the load finishes.

Without a gate, an early request reads an empty or null stream and gets a "not found" response — not a wait. The gate turns that race into "wait until loaded, then respond."

Canonical example: MeshNodeInitGateName, opened by MeshNodeTypeSource once the hub's own MeshNode has been read. Without the gate, a GetDataRequest for MeshNodeReference would race the load and silently return null.

If your data is available synchronously — built-in nodes, WithInitialData([...]), a static provider — you don't need a gate at all. Open it eagerly at config time so no requests are unnecessarily queued. MeshDataSource.WithMeshNodes() does exactly this in its "built-in" and "static" branches.

🚨 The absolute rule: open gates from reactive observables, never from await

The gate opens when a condition is met — concretely, the first emission of an observable, not the completion of a Task. Awaiting a Task to open the gate captures the calling scheduler (typically the grain scheduler). While the await is in-flight, the hub's action block is blocked. While the action block is blocked, every other message — including the queued ones the gate is meant to release — waits behind it. The gate never opens.

Do not write this:

// ❌ DEADLOCK — await captures the grain scheduler; gate never opens.
protected override async Task<InstanceCollection> InitializeAsync(...)
{
    var node = await persistence.GetNode(path).FirstAsync().ToTask(ct);
    hub.OpenGate(MeshNodeInitGateName);   // never reached on the offending path
    return BuildCollection(node);
}

Write this instead:

// ✅ Pure reactive composition. The framework subscribes; the gate opens on emission.
protected override IObservable<InstanceCollection> Initialize(...)
    => persistence.GetNode(path)              // already IObservable
        .FirstAsync()
        .Select(node =>
        {
            hub.OpenGate(MeshNodeInitGateName);
            return BuildCollection(node);
        });

The framework consumes Initialize(...) via .Subscribe(...) (or composes it into a larger observable that bridges to Task only at the outer edge, e.g. inside WithInitialization). The Initialize body itself must contain no await, no .ToTask, and no Observable.FromAsync over a hub round-trip.

Observable.FromAsync is never permitted

🚨 Observable.FromAsync is FORBIDDEN anywhere in src/ — including for a "pure" DB hit, file I/O, an EF query or an HTTP fetch. A bare FromAsync runs the function's synchronous prologue on the subscribing thread (the hub scheduler when the subscribe happens mid-handler) and applies no concurrency bound. The only occurrence in the tree is sealed inside IoPool itself; every other mention in src/ is a comment saying "never".

A genuinely-async leaf goes through IIoPool — pool.Invoke(ct => SomethingAsync(ct)) for a Task<T> leaf, pool.InvokeBlocking(...) for a sync-blocking/CPU leaf — see Controlled I/O Pooling.

It is doubly wrong over a hub round-trip — anything whose completion depends on the hub's own action block making forward progress (a cross-hub request, a workspace.GetRemoteStream(...).Take(1).ToTask()): pooling it does not help, because the thing it waits for can never be dequeued. Compose those reactively and Subscribe. See Asynchronous Calls for the full rule.

The ITypeSource.Initialize contract

ITypeSource exposes a single reactive method:

internal IObservable<InstanceCollection> Initialize(
    WorkspaceReference<InstanceCollection> reference,
    CancellationToken cancellationToken);

Implementations emit exactly one InstanceCollection. The framework's data-source initializer (GenericUnpartitionedDataSource.GetInitialValueAsync and its partitioned twin) composes per-type-source Initialize calls via SelectMany + Aggregate, then bridges to Task<EntityStore> exactly once at the StreamConfiguration<T>.WithInitialization(...) edge (that overload consumes a Func<…, Task<TStream>>; the settle is the single .FirstAsync().ObserveCompletion(...) there — .ToTask() is forbidden, 2026-08-30).

ITypeSource.Initialize is IObservable<InstanceCollection> — implementations compose reactively; there is no Task-returning surface to await.

Opening a gate from inside Initialize

The cleanest approach is to open the gate inside the .Select that produces the InstanceCollection:

protected override IObservable<InstanceCollection> Initialize(
    WorkspaceReference<InstanceCollection> reference,
    CancellationToken cancellationToken)
    => storage.Get(...)                   // IObservable<...>
        .FirstAsync()
        .Select(loaded =>
        {
            // 1. Side effect: gate open. Idempotent — safe on re-emission.
            workspace.Hub.OpenGate(MyGateName);

            // 2. Pure result.
            return new InstanceCollection(loaded, TypeDefinition.GetKey);
        });

Why .Select rather than .Do? Placing the side effect inside .Select ties it to the same emission as the result. If .Do fires before a downstream .Where filter drops the emission, the gate opens while the InstanceCollection is never delivered — a subtle divergence. .Select keeps the side effect and the result atomic.

Handling load failures

If the load can fail, open the gate in the catch branch too — using an empty InstanceCollection so the hub serves "not found" rather than infinite silence:

.Catch<InstanceCollection, Exception>(ex =>
{
    logger.LogError(ex, "Init failed for {HubPath} — opening gate with empty collection so traffic doesn't stall", hubPath);
    workspace.Hub.OpenGate(MyGateName);
    return Observable.Return(new InstanceCollection(System.Array.Empty<object>(), TypeDefinition.GetKey));
});

Without this, a failed load leaves the hub permanently gated and every queued message eventually times out with no useful error surfaced.

Opening a gate from a hub-init hook

When the gate condition isn't "type source loaded its collection" — for example, "a remote readiness signal arrived" or "the permission service has refreshed" — register the gate in hub config and open it from a WithInitialization(...) hook:

config
    .WithInitializationGate(MyGateName, d => d.Message is BootstrapRequest)
    .WithInitialization(hub =>
        // Return the readiness observable — the framework composes the BuildupActions with
        // Observable.Concat and opens the framework Initialize gate when they complete.
        // The side effect (OpenGate) rides the emission; NEVER `await`-bridge here.
        readinessObservable
            .Take(1)
            .Select(_ => { hub.OpenGate(MyGateName); return Unit.Default; })
            .Catch<Unit, Exception>(ex =>
            {
                logger.LogError(ex, "Readiness signal failed; opening gate to release queued traffic");
                hub.OpenGate(MyGateName);
                return Observable.Return(Unit.Default);
            }));

MessageHubConfiguration has exactly two WithInitialization overloads, and neither returns a Task:

There is no Func<IMessageHub, CancellationToken, Task> overload — return Task.CompletedTask; does not compile here. (The Task-returning WithInitialization you may have seen belongs to StreamConfiguration<T>, a different type, used by the data-source layer.)

A hang inside the observable overload is bounded: HandleInitialize wraps the composed Concat in .Timeout(Configuration.StartupTimeout ?? 120s) and, on fault or timeout, puts the hub in a FAILED state and opens the framework gate anyway so rejections can flow — see Hub Initialization Failure. Awaiting inside this hook is still the same deadlock as awaiting inside Initialize.

🚫 What a gate is NOT for

A gate is a startup mechanism. Two properties decide everything it can and cannot do:

1. It defers INBOUND deliveries — messages arriving at this hub. It has no effect on work the hub originates (an IObservable subscription, a timer, a reconcile loop).
2. It is ONE-WAY. OpenGate is idempotent and there is no CloseGate. A gate models "not ready yet → ready", once, per hub lifetime.

So a gate cannot throttle, debounce, or suppress anything in steady state. If a hub writes nodes in a loop, a gate will not stop it — the gate opened long before, and the writes were never inbound messages in the first place.

The real fix for a write loop is to stop feeding a reconcile its own output. A reconcile driven by a query over a subtree that CONTAINS the nodes it writes re-triggers itself on every write; if any predicate in it fails to match, that is an unbounded write storm rather than one wasted write. Filter the trigger so the reconcile's own bookkeeping nodes cannot schedule another pass. (Live example: a plugin's _Policy reached version 257,000 at ~14 writes/minute because one predicate could never match — MeshWeaver.Plugins#223.)

⚠️ The gate that defers its own opener

Before gating a hub on work that writes nodes, check where the responses land.

IMeshService is constructed with the calling hub (MeshService(… IMessageHub hub), resolved from that hub's service provider). So mesh.CreateNode(...) / CreateOrUpdateNode(...) / DeleteNode(...) post a request and receive the response back on the hub that called them. A gate whose opening condition is "my seeding finished" therefore defers the very response that completes the seeding:

// ❌ DEADLOCK — the write's response is inbound traffic, and this gate is holding it.
config
    .WithInitializationGate(SeedGateName)                      // no bypass for the response
    .WithInitialization(hub => { Seed(hub).Subscribe(_ => hub.OpenGate(SeedGateName)); … });

Symptom: not an error — a hub where every message waits the full 30 s DeferralTimeout, then fails. Same shape as the InitializeHubRequest incident above, and it is why that message is now bypassed unconditionally.

If you do need such a gate, the bypass predicate must let through the replies the work depends on (CreateNodeResponse, CreateOrUpdateNodeResponse, DeleteNodeResponse, plus the data traffic of any query it awaits) — and that set is worth pinning with a test, because a missing entry wedges the hub only on the path that actually writes.

Gate-bypass predicates

A bypass predicate returns true for messages whose handling cannot wait. Anything user-driven — GetDataRequest, queries, mutations — should not bypass: that is the entire reason the gate exists.

The most common bypass case is CreateNodeRequest: the hub is being asked to come into existence, so deferring would prevent the gate from ever mattering.

.WithInitializationGate(MeshNodeInitGateName, d => d.Message is CreateNodeRequest)

Messages the framework always bypasses

The framework unconditionally bypasses all gates for the following system messages — see MessageService.cs (the delivery.Message is ShutdownRequest or … short-circuit evaluated before any gate predicate). There is no need to repeat these in your own predicate.

Background: the InitializeHubRequest bypass was added after a prod incident where thread-hub SubscribeRequests timed out at 30 s because InitializeHubRequest was sitting queued behind MeshNodeInitGateName.

Anti-patterns

// ❌ Awaiting inside the init function — captures the calling scheduler.
//    The gate never opens because nothing else can run on this scheduler
//    while the await is pending.
protected override async Task<InstanceCollection> InitializeAsync(...)
{
    var x = await SomeHubRoundTrip();       // deadlock
    hub.OpenGate(name);
    return ...;
}

// ❌ Opening the gate eagerly before the data is loaded.
//    Defeats the purpose: queued reads now run against an empty collection.
config.WithInitializationGate(name, d => d.Message is CreateNodeRequest)
      .WithInitialization(hub => hub.OpenGate(name));   // the Action<IMessageHub> overload

// ❌ Observable.FromAsync at all — forbidden outside IoPool, and over a hub
//    round-trip it is the same deadlock as pattern one, hidden one level deeper.
.WithInitialization(hub =>
    Observable.FromAsync(ct => SomeHubAwait(ct))
        .Select(_ => { hub.OpenGate(name); return Unit.Default; }));

// ❌ No error handler — gate stays closed forever on load failure.
//    Every queued message eventually times out with no useful error surfaced.
ownStream.Select(...).Subscribe(v => hub.OpenGate(name) /* no OnError */);

Naming and organisation

• Define gate names as public const string on the module that owns the gate. Cross-file string typos are a silent foot-gun.
• One gate per readiness condition. Multiple independent loads can use multiple stacked gates; opening one does not release messages waiting on another.
• Document the constant: state what condition opens the gate, which bypass predicates apply, and which sources are idempotent-safe.

public static class MeshNodeExtensions
{
    /// <summary>
    /// Gate name for "the per-hub MeshNode collection has been loaded from
    /// persistence". Bypasses CreateNodeRequest. Opened by
    /// <c>MeshNodeTypeSource</c> once the own-node emission is accepted (and by
    /// <c>MeshDataSource</c> on the built-in / static branches). Idempotent — safe to
    /// open from multiple branches (load success, load failure, hubs without a
    /// persisted node).
    /// </summary>
    public const string MeshNodeInitGateName = "MeshNodeInit";
}

Related

• Asynchronous Calls in MeshWeaver — the broader rule: no await of hub round-trips anywhere in mesh-reachable code.
• CQRS — Queries vs. Content Access — queries are lagged and not the right primitive for known-path content reads, regardless of gating.
• src/MeshWeaver.Graph/MeshNodeTypeSource.cs — canonical reactive Initialize implementation that opens MeshNodeInitGateName.
• src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs — GetInitialValueAsync shows the framework-edge bridge: per-type-source Initialize calls composed via SelectMany + Aggregate, with a single .ToTask at the StreamConfiguration<T>.WithInitialization boundary.
• src/MeshWeaver.Messaging.Hub/MessageHub.cs — HandleInitialize, the reactive BuildupActions composition with its StartupTimeout liveness bound.