Operations Framework: Multi-Host Invalidation, CQRS, and Reliable Command Processing

The Operations Framework (OF) provides a robust foundation for building distributed systems with Fusion. It solves several critical challenges that arise when running multiple instances of an application:

• Multi-host cache invalidation: When data changes on one server, all other servers must invalidate their cached computed values
• Reliable command processing: Commands must be executed exactly once, even in the face of failures
• Event-driven architecture: Commands can produce events that are processed asynchronously with guaranteed delivery

Why Do You Need Operations Framework?

Consider a typical multi-server deployment:

When a user on Server A updates their profile:

1. Server A writes to the database and invalidates its local cache
2. Servers B and C still have stale data in their caches
3. Users connected to B and C see outdated information

Without Operations Framework, you'd have to implement:

• A message queue or pub/sub system for cross-server notifications
• Retry logic for failed operations
• Deduplication to prevent processing the same operation twice
• Transaction handling to ensure atomicity

With Operations Framework, all of this is handled automatically.

Required Packages

Package	Purpose
ActualLab.Fusion	Core OF abstractions and in-memory implementation
ActualLab.Fusion.EntityFramework	EF Core implementation: DbOperationScope, operation logging, DbContext integration
ActualLab.Fusion.EntityFramework.Npgsql	PostgreSQL: NpgsqlOperationLogWatcher for LISTEN/NOTIFY
ActualLab.Fusion.EntityFramework.Redis	Redis: RedisOperationLogWatcher for pub/sub notifications

TIP

The base ActualLab.Fusion.EntityFramework package includes FileSystemOperationLogWatcher which works with any database but requires shared filesystem access. For production multi-host deployments, use database-specific watchers (Npgsql) or Redis.

The Outbox Pattern

Operations Framework implements the Transactional Outbox Pattern – a well-known solution for reliable messaging in distributed systems.

The Problem

In distributed systems, you often need to:

1. Update your database
2. Publish a message/event to notify other services

But what if step 2 fails after step 1 succeeds? You have inconsistent state.

The Solution: Outbox Pattern

Instead of publishing directly, write the message to an "outbox" table in the same transaction as your business data:

This guarantees at-least-once delivery: if the transaction commits, the operation will eventually be processed. If it fails, nothing is written.

How OF Implements It

1. DbOperationScope wraps your command in a database transaction
2. DbOperation entity stores the operation in the same transaction
3. DbOperationLogReader (background service) watches for new operations
4. Operation Log Watchers provide instant notifications (PostgreSQL NOTIFY, Redis Pub/Sub, etc.)
5. OperationCompletionNotifier triggers invalidation on all hosts

Core Concepts

Operation

An Operation represents an action that can be logged and replayed. Currently, only commands act as operations, but the framework is designed to support other types in the future.

Key properties:

• Uuid – Unique identifier
• HostId – The server that executed the operation
• Command – The command that was executed
• Items – Data passed between execution and invalidation phases
• NestedOperations – Child operations executed during this operation
• Events – Events produced by this operation

Operation Scope

An Operation Scope provides the context for operation execution:

• DbOperationScope: Persistent operations stored in database (default for database commands)
• InMemoryOperationScope: Transient operations that don't persist (for in-memory commands)

Invalidation Mode

When an operation is "replayed" on other hosts, it runs in invalidation mode:

• The command handler's main logic is skipped
• Only the invalidation block executes
• This ensures all hosts invalidate the same computed values

Quick Start

1. Add DbSet for Operations

public DbSet<DbOperation> Operations { get; protected set; } = null!;
public DbSet<DbEvent> Events { get; protected set; } = null!;

2. Configure Services

public static void ConfigureServices(IServiceCollection services, IHostEnvironment Env)
{
    services.AddDbContextServices<AppDbContext>(db => {
        // Uncomment if you'll be using AddRedisOperationLogWatcher
        // db.AddRedisDb("127.0.0.1", "FusionDocumentation.PartO");

        db.AddOperations(operations => {
            // This call enabled Operations Framework (OF) for AppDbContext.
            operations.ConfigureOperationLogReader(_ => new() {
                // We use AddFileSystemOperationLogWatcher, so unconditional wake up period
                // can be arbitrary long – all depends on the reliability of Notifier-Monitor chain.
                // See what .ToRandom does – most of timeouts in Fusion settings are RandomTimeSpan-s,
                // but you can provide a normal one too – there is an implicit conversion from it.
                CheckPeriod = TimeSpan.FromSeconds(Env.IsDevelopment() ? 60 : 5).ToRandom(0.05),
            });
            // Optionally enable file-based operation log watcher
            operations.AddFileSystemOperationLogWatcher();

            // Or, if you use PostgreSQL, use this instead of above line
            // operations.AddNpgsqlOperationLogWatcher();

            // Or, if you use Redis, use this instead of above line
            // operations.AddRedisOperationLogWatcher();
        });
    });
}

Note: OF works solely on the server side, so you don't need similar configuration in your Blazor WebAssembly client.

3. Create Command and Handler

public record PostMessageCommand(Session Session, string Text) : ICommand<ChatMessage>;

[CommandHandler]
public virtual async Task<ChatMessage> PostMessage(
    PostMessageCommand command, CancellationToken cancellationToken = default)
{
    if (Invalidation.IsActive) {
        _ = PseudoGetAnyChatTail();
        return default!;
    }

    await using var dbContext = await DbHub.CreateOperationDbContext(cancellationToken);
    // Actual code...
    var message = await PostMessageImpl(dbContext, command, cancellationToken);
    return message;
}

Command Handler Structure

A command handler with Operations Framework follows this pattern:

[CommandHandler]
public virtual async Task<TResult> HandleCommand(
    TCommand command, CancellationToken cancellationToken = default)
{
    // 1. INVALIDATION BLOCK - runs on ALL hosts after successful execution
    if (Invalidation.IsActive) {
        // Invalidate computed values that depend on the data being changed
        _ = GetSomeData(command.Id, default);
        _ = GetRelatedData(command.RelatedId, default);
        return default!;  // Return value is ignored in invalidation mode
    }

    // 2. MAIN LOGIC - runs only on the originating host
    await using var dbContext = await DbHub.CreateOperationDbContext(cancellationToken);

    // Perform your business logic
    var result = await DoWork(dbContext, command, cancellationToken);

    await dbContext.SaveChangesAsync(cancellationToken);
    return result;
}

Key Points

1. virtual modifier – Required for Fusion's proxy generation
2. [CommandHandler] attribute – Registers this method as a command handler
3. Invalidation.IsActive check – First thing in the method
4. CreateOperationDbContext – Creates a DbContext that participates in the operation scope

Don't spawn background work from inside an invalidation block

The invalidation block (and any synchronous Invalidated event handler) runs inside an Invalidation.Begin() scope. That scope is AsyncLocal-based, so if you start a Task.Run or similar background work from within it, the spawned work inherits the same ambient invalidation context even after the block exits. Any compute method it calls will silently skip computation and invalidate instead – see Computed.BeginIsolation() for the guardrail.

Passing Data to Invalidation Block

The invalidation block runs on all hosts, but the main logic only runs on the originating host. To pass data from main logic to invalidation, use Operation.Items:

public virtual async Task SignOut(
    SignOutCommand command, CancellationToken cancellationToken = default)
{
    // ...
    var context = CommandContext.GetCurrent();
    if (Invalidation.IsActive) {
        // Fetch operation item
        var invSessionInfo = context.Operation.Items.KeylessGet<SessionInfo>();
        if (invSessionInfo is not null) {
            // Use it
            _ = GetUser(invSessionInfo.UserId, default);
            _ = GetUserSessions(invSessionInfo.UserId, default);
        }
        return;
    }

    await using var dbContext = await DbHub.CreateOperationDbContext(cancellationToken).ConfigureAwait(false);

    var dbSessionInfo = await Sessions.FindOrCreate(dbContext, command.Session, cancellationToken).ConfigureAwait(false);
    var sessionInfo = dbSessionInfo.ToModel();
    if (sessionInfo.IsSignOutForced)
        return;

    // Store operation item for invalidation logic
    context.Operation.Items.KeylessSet(sessionInfo);
    // ...
}

How It Works

1. During execution: Store data with context.Operation.Items.KeylessSet(value)
2. During invalidation: Retrieve data with context.Operation.Items.KeylessGet<T>()
3. Serialization: Items are JSON-serialized and stored with the operation in the database

Note: Operation.Items differs from CommandContext.Items:

• CommandContext.Items exists only during command execution on the originating host
• Operation.Items is persisted and available on all hosts during invalidation

Testing Invalidation

Dependency propagation invalidates every transitive dependant of a call automatically, but it can't invent an invalidation for a directly affected call that a handler's invalidation block forgot to enumerate. Nothing catches that at compile time – a query added later, a changed data dependency, or a broad aggregate query that isn't obviously "related" to the command can silently fall out of sync with reality.

The mitigation isn't an analyzer (the design space is too broad for one to pay off yet) – it's a test convention: for every mutating command, write a test that proves it invalidates both the entity-specific query it targets and any aggregate query whose result depends on the same data.

public record KeyValueService_Set(string Key, string Value) : ICommand<Unit>;

public class KeyValueService : IComputeService
{
    private readonly ConcurrentDictionary<string, string> _values = new();

    // Entity-specific query
    [ComputeMethod]
    public virtual Task<string?> Get(string key, CancellationToken cancellationToken = default)
        => Task.FromResult(_values.GetValueOrDefault(key));

    // Aggregate query
    [ComputeMethod]
    public virtual Task<int> Count(CancellationToken cancellationToken = default)
        => Task.FromResult(_values.Count);

    [CommandHandler]
    public virtual Task<Unit> Set(KeyValueService_Set command, CancellationToken cancellationToken = default)
    {
        if (Invalidation.IsActive) {
            // Every mutating command handler must invalidate BOTH the entity-specific
            // query it directly affects AND every aggregate query whose result may change --
            // dependency tracking alone won't discover an omitted root call.
            _ = Get(command.Key, default);
            _ = Count(default);
            return Task.FromResult(Unit.Default);
        }

        // Requests an operation scope so this in-memory command gets completion
        // notifications (and therefore an invalidation replay) -- see InMemoryOperationScopeProvider
        InMemoryOperationScope.Require();
        _values[command.Key] = command.Value;
        return Task.FromResult(Unit.Default);
    }
}

Capture both computed values before calling the command, then assert both got invalidated:

var services = new ServiceCollection();
services.AddFusion().AddComputeService<KeyValueService>();
var sp = services.BuildServiceProvider();
var commander = sp.Commander();
var kv = sp.GetRequiredService<KeyValueService>();

// Capture both queries BEFORE the mutating command runs
var cGet = await Computed.Capture(() => kv.Get("a"));
var cCount = await Computed.Capture(() => kv.Count());
WriteLine($"Before Set: Get.IsConsistent={cGet.IsConsistent()}, Count.IsConsistent={cCount.IsConsistent()}");

await commander.Call(new KeyValueService_Set("a", "1"));

// A test would assert both are False here -- if Set() only invalidated
// Get(), this would (correctly) fail and catch the missing Count() invalidation
WriteLine($"After Set:  Get.IsConsistent={cGet.IsConsistent()}, Count.IsConsistent={cCount.IsConsistent()}");

The output:

Before Set: Get.IsConsistent=True, Count.IsConsistent=True
After Set:  Get.IsConsistent=False, Count.IsConsistent=False

If Set's invalidation block only called Get(command.Key, default) and forgot Count(default), this test would catch it immediately: Count.IsConsistent would stay True after the command completes. Without the test, that gap would surface later as a UI aggregate (a count, a list, a total) that never updates, with nothing in the logs pointing at the cause.

Nested Commands

When one command calls another, the nested command is automatically logged and its invalidation logic runs on all hosts:

[CommandHandler]
public virtual async Task<Order> CreateOrder(
    CreateOrderCommand command, CancellationToken cancellationToken = default)
{
    if (Invalidation.IsActive) {
        _ = GetOrder(command.OrderId, default);
        return default!;
    }

    await using var dbContext = await DbHub.CreateOperationDbContext(cancellationToken);

    var order = new Order { /* ... */ };
    dbContext.Orders.Add(order);
    await dbContext.SaveChangesAsync(cancellationToken);

    // This nested command is automatically logged
    await Commander.Call(new SendOrderConfirmationCommand(order.Id), cancellationToken);

    return order;
}

The nested command's Operation.Items are captured independently, so there's no collision with the parent command's items.

Command Pipeline

Operations Framework adds several filtering handlers to the command pipeline:

Priority	Handler	Purpose
100,000	OperationReprocessor	Retries commands that fail with transient errors
11,000	NestedOperationLogger	Logs nested commands and their items
10,000	InMemoryOperationScopeProvider	Provides transient scope, runs completion
1,000	DbOperationScopeProvider	Provides database scope for each DbContext type
100	InvalidatingCommandCompletionHandler	Runs invalidation for completed operations

Invariants and Guarantees

Invalidation replay and completion listeners must not fail

InvalidatingCommandCompletionHandler.TryInvalidate and OperationCompletionNotifier's dispatch to IOperationCompletionListeners both catch every exception, log it, and otherwise treat the operation as fully processed – a failed invalidation replay or a failed listener is not retried, and (on the originating host) it's never revisited within the same process lifetime.

This is a deliberate design trade-off, but it rests on a hard, mostly-unenforced contract: invalidation logic and completion listeners must not fail. Concretely, that means:

• Invalidation blocks (the if (Invalidation.IsActive) { ... } branch) and any compute method they transitively call must be synchronous or otherwise guaranteed to complete, side-effect-free, and independent of failure-prone infrastructure (no I/O that can legitimately fail).
• Custom IOperationCompletionListener implementations must not throw under any input they can reasonably observe.

If this contract is violated, the swallowed exception is your only signal, and the corresponding invalidation (or completion notification) is simply lost for that operation. When writing either kind of code, test it explicitly against failure scenarios you'd otherwise rely on retry semantics to paper over – retries are not coming.

Completion listener delivery is at-least-once

For operations delivered by the log reader (i.e. everything except the originating host's in-process notification), IOperationCompletionListener dispatch is at-least-once, not exactly-once. When a listener fails, the reader redelivers the whole operation; on a crash–restart it also re-reads every operation still within StartOffset. Redelivery re-runs all listeners registered for that operation – including the ones that already succeeded on the previous attempt, since the reader tracks progress per operation, not per listener. Listeners with external side effects must therefore be idempotent, keyed off Operation.Uuid (or an equivalent per-event key) so a re-run is a no-op.

Command completion isn't a cluster-wide freshness boundary

After a command's transaction commits, invalidation still has to pass through local completion handling, watcher notification (or the poll fallback), operation replay on every other host, and – for RPC clients – a further Invalidate message. During that interval, another host or a connected client can legitimately observe a cached pre-command value. This is intentional eventual consistency, not a bug: Commander.Call returning does not mean every dependent cache cluster-wide has been invalidated.

If a caller needs cross-host read-your-write semantics (e.g., "the very next read on any host must see this write"), command completion alone doesn't provide it – that requires an explicit version or synchronization mechanism layered on top (e.g., threading a version/timestamp through the response and having the reader wait for a computed at least that fresh).

Backend Commands

Commands that should only execute on the server should implement IBackendCommand:

public record DeleteUserCommand(UserId UserId) : ICommand<Unit>, IBackendCommand;

This ensures:

• The command can only be processed by backend servers
• Client-side proxies won't attempt to handle it
• RPC layer enforces server-side execution

Further Reading

• Events – Producing and consuming events from operations
• Transient Operations and Reprocessing – In-memory operations and retry logic
• Configuration Options – All configuration options explained
• Log Watchers – PostgreSQL, Redis, FileSystem log watchers
• Diagrams – Visual representations of OF internals
• Cheat Sheet – Quick reference

Learning More

To explore OF's internals, check out:

• DbContextBuilder.AddOperations
• FusionBuilder

HostId

HostId identifies each process in your cluster. It includes:

• Machine name
• Unique process ID
• Unique ID per IoC container (useful for testing)

This allows OF to determine if an operation originated locally or from a peer.

InvalidatingCommandCompletionHandler

The logic that determines whether a command requires invalidation is in InvalidatingCommandCompletionHandler.IsRequired(). It returns true for any command with a final handler whose service implements IComputeService, but not for compute service clients (when RpcServiceMode.Client is set).

Getting Help

If you run into issues, join Fusion Place and ask questions. The author (Alex Y.) is active and happy to help.