Resrcify.SharedKernel.MessageBus 4.0.0

Resrcify.SharedKernel.MessageBus

Resrcify.SharedKernel.MessageBus is the message bus between services, on Rebus and RabbitMQ.

It replaces MassTransit with:

• Scatter-gather (request/response):
  • Requesting: an IScatterGatherHandler<TEvent, TRequest, TResponse> handles a domain event.
    1. It returns one request per item.
    2. The requests are sent and the replies collected in memory.
    3. It applies everything that came back in one go.
  • Responding: one queue per request type, each under its own rate limit per instance.
• Publish/subscribe (integration events): IEventBus.PublishAsync in one service, an IIntegrationEventHandler<TEvent> in each service that wants it.

Moving a service from MassTransit: the migration guide in the Resrcify.SwgohApi repository (docs/migrating-from-masstransit.md).

It keeps no state and adds no tables. Durability comes from the service's existing outbox. The event commits with the aggregate that raised it, and the outbox marks it processed only after the replies have been gathered and applied. If anything fails, the outbox retries the whole event. These events run in the outbox's scatter-gather lane (see the UnitOfWork README), so a waiting batch never holds up other outbox messages.

Table of Contents

• Resrcify.SharedKernel.MessageBus
  • Table of Contents
  • Where each piece lives
  • Install
  • Message names
  • Requesting: scatter-gather
  • Requesting inside a request: gather before the handler
  • Asking directly: IScatterGatherClient
  • Failures: the result pattern
  • Responding: rate-limited queues
  • Publish/subscribe: integration events
  • The error queue: show, delete, move back
  • Running on a RabbitMQ cluster
  • Running without RabbitMQ: in memory
  • Checking that two copies of a message agree
  • Large messages: compression
  • Customising: strategies
  • Health check
  • Tracing and metrics
  • Behaviour you can rely on

Where each piece lives

Handlers only see the contracts, so the Application layer never references Rebus or this package. The ArchitectureTesting package's ConventionalLayerDependencyTests enforces it (see the root README).

Inside the package

Install

<PackageReference Include="Resrcify.SharedKernel.MessageBus" Version="<latest>" />

It brings Rebus, Rebus.RabbitMq, Rebus.ServiceProvider, Rebus.Diagnostics and the health-check and hosting abstractions with it. It reaches the outbox only through its contracts (IOutboxLaneEvent, IOutboxMessageContext in Resrcify.SharedKernel.Abstractions), so a service that only publishes or answers needs no outbox. Scatter-gather handlers, which wait through the outbox, need Resrcify.SharedKernel.UnitOfWork (AddOutboxProcessing) as well.

Message names

A message goes by its class name on the wire, without the namespace or assembly. Nothing has to be named: the requester's and the responder's classes (or the publisher's and the subscribers') only need the same name, and can live in different namespaces and packages. A request's queue is named after its request class, on both sides.

When one side's class has another name, bind it to the other side's: AddMessage<GuildPayload>(nameof(GetGuildRequest)) (nameof keeps it compiler-checked). For a breaking change, add a class with a new name (ShardRankChangedV2) and run both while the other side moves. If two message types this service receives share a name, the first message fails with an error that names both.

Prefix event classes with their owner (ShardRankChanged, DiscordDeliveryFailed). A request goes to its owner's queue, but an event's name is global on the broker: two services publishing events of the same name would reach every subscriber of either. Each event carries its publisher's service name (its input queue, or WithServiceName("shard")), and a subscriber that sees one event name arrive from two services logs a warning and counts it in messagebus.events.name_clashes.

Requesting: scatter-gather

In the requesting service (e.g. ShardManagement). It says what it asks for; it has no say over how fast it is answered.

// Infrastructure: once per service.
// using Resrcify.SharedKernel.MessageBus.Extensions;     (AddMessageBus)
// using Resrcify.SharedKernel.MessageBus.Configuration;  (RabbitMqConnection, MessageBusBuilder)
services.AddMessageBus(bus => bus
    .UseRabbitMq(provider =>
    {
        var options = provider.GetRequiredService<IOptions<MessageBusOptions>>().Value;   // the service's own options
        return new RabbitMqConnection(options.Host, options.Port, options.Username, options.Password);
    })
    .AddRequest<PlayerArenaProfileRequest, PlayerArenaProfileResponse>()   // sent to the PlayerArenaProfileRequest queue
    .AddScatterGather());

A virtual host other than /, or TLS, goes on the connection (new RabbitMqConnection(...) { VirtualHost = "svc", UseTls = true }), so every connection the package makes uses it: each bus', and the broker watcher's that the health check and the restarts after an outage rely on. Certificates and other connection-factory settings go in the configuration strategy, in both ConfigureTransport (the buses) and ConfigureConnectionFactory (the watcher).

AddScatterGather opens a reply queue private to this service instance (deleted when the instance goes away), so with several instances each one receives only its own replies. Handlers need no registration: it follows the mediator's assembly scan (whether AddMediator runs before or after it), wires each handler as its event's notification handler, and puts their events in the outbox's scatter-gather lane. Handlers outside the mediator's assemblies can be added with AddScatterGather(extraAssembly).

The aggregate raises an ordinary domain event. Make it IDedupable to keep one in flight per key:

public sealed record ShardRankSyncRequested(Guid Id, Guid ShardId) : DomainEvent(Id), IDedupable
{
    public string DedupKey => ShardId.ToString("N");
}

The handler builds the requests, then applies the gathered replies. Don't save in it; the outbox commits.

internal sealed class SyncShardRanksHandler(IShardRepository shards)
    : IScatterGatherHandler<ShardRankSyncRequested, PlayerArenaProfileRequest, PlayerArenaProfileResponse>
{
    public TimeSpan Timeout => TimeSpan.FromSeconds(50);

    public async Task<IReadOnlyDictionary<string, PlayerArenaProfileRequest>> ScatterAsync(
        ShardRankSyncRequested notification, CancellationToken cancellationToken)
    {
        var shard = await shards.GetShardWithMembers(notification.ShardId, cancellationToken);
        return shard!.ShardMembers.ToDictionary(m => m.Id.ToString("N"), m => new PlayerArenaProfileRequest { ... });
    }

    public async Task GatherAsync(
        ShardRankSyncRequested notification, IGathered<PlayerArenaProfileResponse> replies, CancellationToken cancellationToken)
    {
        // replies.Results (by key), replies.Failures (each item's errors), replies.UnansweredKeys,
        // replies.SettledKeys (a definite answer: a result, or a failure that is the request's fault),
        // or one item as a Result<PlayerArenaProfileResponse>: replies[key]
        var shard = await shards.GetShardWithMembers(notification.ShardId, cancellationToken);
        shard!.SyncShard(...);   // all or nothing: only when every member is in replies.SettledKeys
    }
}

Publishing the event straight through IPublisher runs the same handler in-process, without the outbox's durability.

Requesting inside a request: gather before the handler

When a request (a query, say) needs answers from another service before it can be handled, its handler can scatter and gather instead. It is sent like any request (ISender.Send), so pipeline behaviors run first; the bus asks, gathers, and hands the replies to the handler. AddScatterGather finds these handlers the same way.

internal sealed class GetGuildRanksQueryHandler
    : IScatterGatherRequestHandler<GetGuildRanksQuery, PlayerArenaProfileRequest, PlayerArenaProfileResponse, Result<GuildRanks>>
{
    public TimeSpan Timeout => TimeSpan.FromSeconds(10);

    public Task<IReadOnlyDictionary<string, PlayerArenaProfileRequest>> ScatterAsync(
        GetGuildRanksQuery request, CancellationToken cancellationToken)
        => ...;   // one request per key

    public Task<Result<GuildRanks>> HandleAsync(
        GetGuildRanksQuery request, IGathered<PlayerArenaProfileResponse> replies, CancellationToken cancellationToken)
        => ...;   // everything that came back, or what came before the timeout
}

IStreamingScatterGatherRequestHandler gets the replies as they arrive (IAsyncEnumerable<IScatterReply<T>>), so it can start on the first one, or stop waiting by stopping the enumeration (e.g. "the first 5 that answer").

The caller waits up to Timeout, and nothing survives a crash. For work that must, raise a domain event and use an IScatterGatherHandler (through the outbox) instead.

Asking directly: IScatterGatherClient

Code that isn't a scatter-gather handler (a query handler, a job) can ask through IScatterGatherClient, which AddScatterGather registers. It replaces MassTransit's IRequestClient<T>:

internal sealed class GuildProfileService(IScatterGatherClient client)
{
    public Task<Result<GetGuildResponse>> GetAsync(GetGuildRequest request, CancellationToken cancellationToken)
        => client.RequestAsync<GetGuildRequest, GetGuildResponse>(request, TimeSpan.FromSeconds(30), cancellationToken);
}

RequestAsync gives the response, the responder's errors, or ScatterGather.Unanswered (ErrorType.Timeout). GatherAsync and StreamAsync do the same for a batch. Nothing survives a crash: for work that must, use an IScatterGatherHandler through the outbox.

Failures: the result pattern

A responder answers with a Result<TResponse> (Resrcify.SharedKernel.Results.Primitives), and the requester gets each item as one:

A failure is an answer: it settles the item without waiting for the timeout. Type its errors for what they mean: "no result" (e.g. the player doesn't exist) is NotFound, answered at once; an upstream that is down stays a Failure, so it is tried again. Change the rule per queue with queue.AnswerFailure = errors => ....

A requester that must not mistake an outage for "no result" uses replies.SettledKeys: the items with a definite answer (a result, or a failure with no IsTransient() error), leaving out those never answered and those whose responder gave up after its last try. An item that answers with a message of the wrong type fails with ScatterGather.UnexpectedReply.

A responder's HandleAsync gets a cancellation token. Pass it on (e.g. to the HTTP call upstream):

• It fires when the requester stops waiting (the request's time-to-live, which is the batch's Timeout). The request is dropped unanswered instead of spending the upstream's budget on a reply nobody reads.
• It fires when the instance stops consuming (its health check turned unhealthy, or it shuts down). The request goes back to the queue, and another instance answers it at once instead of after the handler finishes.

Responding: rate-limited queues

In the providing service (e.g. SwgohApi). The provider owns the queue and its limits: it knows what its upstream allows, and every requester shares the same queue, so the limit holds however many services ask.

services.AddHealthChecks().AddCheck<GalaxyOfHeroesServiceHealthCheck>("galaxy-of-heroes-api", tags: ["ready", "game"]);
services.AddMessageBus(bus => bus
    .UseRabbitMq(...)
    // Consumes the PlayerArenaProfileRequest queue. Gate it on a tag only the upstream's check carries.
    .AddRateLimitedQueue<PlayerArenaProfileRequest, PlayerArenaProfileResponse, PlayerArenaProfileHandler>(
        queue => { queue.PerSecond = 90; queue.Burst = 5; queue.HealthCheckTag = "game"; }));

internal sealed class PlayerArenaProfileHandler(ISender sender)
    : IRequestResponder<PlayerArenaProfileRequest, PlayerArenaProfileResponse>
{
    public async Task<Result<PlayerArenaProfileResponse>> HandleAsync(PlayerArenaProfileRequest request, CancellationToken ct)
        => ...;   // NotFound/Validation/...: answered at once; any other failure: tried again, then answered
}

Each AddRateLimitedQueue runs its own bus and token bucket, so every request type has its own limit per instance. Instances compete on the same queue, so adding one adds capacity.

In any one second at most PerSecond + Burst requests start. Tokens come a refill step at a time (5 at 90/s, every 55.6 ms), and the calls of one step reach the upstream a little spread, which can shift one step into the next second there. Leave that step as headroom: for an upstream that allows 100 a second, 90 + burst 5.

Without a responder class: through the mediator

Most responders only turn the request into a mediator request and send it. That is the default:

services.AddMessageBus(bus => bus
    .UseRabbitMq(...)
    .AddRateLimitedQueue<GetGuildRequest, GetGuildResponse>(
        request => new GetGuild(request.GuildId),          // an IRequest<Result<GetGuildResponse>>
        queue => { queue.PerSecond = 18; queue.HealthCheckTag = "game"; }));

The mediator's handler (with its pipeline behaviors) answers, and its result is treated like a responder's (see Failures: the result pattern).

Publish/subscribe: integration events

An integration event goes to every service that subscribes, and within a service to one of its instances. Publisher and subscribers agree on the event's class name, not its namespace, so each side keeps its own class.

Publishing. AddMessageBus registers IEventBus (no input queue needed). Publish from a domain event handler, so the outbox runs it: the change and its domain event commit together, and the publish is retried until it succeeds.

services.AddMessageBus(bus => bus.UseRabbitMq(...));   // nothing to register for publishing

internal sealed class RankChangedEventHandler(IEventBus eventBus) : IDomainEventHandler<RankChangedEvent>
{
    public Task Handle(RankChangedEvent notification, CancellationToken cancellationToken)
        => eventBus.PublishAsync(new RankChangedEventMessage(...), cancellationToken);
}

Subscribing. Give the service an input queue (its instances share it) and AddEventHandlers(). Every IIntegrationEventHandler<TEvent> in the assemblies the mediator scans is found, and the input queue subscribes to its event when the service starts.

services.AddMessageBus(bus => bus
    .UseRabbitMq(...)
    .WithInputQueue("discordmanagement")
    .AddEventHandlers()   // subscribes to every event with an IIntegrationEventHandler
    // Optional: one at a time per shard, like MassTransit's UsePartitioner(16, e => e.ShardId).
    // In publish order per shard; the service's input queue then has a single active consumer (one instance
    // consumes, the others stand by). See HandleInPartitions for retries and the error queue.
    .ConfigureSubscription<PayoutRotatedEventMessage>(options => options.HandleInPartitions(e => e.ShardId)));

internal sealed class RankChangedEventMessageHandler(ISender sender) : IIntegrationEventHandler<RankChangedEventMessage>
{
    public Task<Result> HandleAsync(RankChangedEventMessage integrationEvent, CancellationToken cancellationToken)
        => sender.Send(new NotifyRankChangedCommand(...), cancellationToken);
}

Forwarding to the mediator. A handler that only turns the event into a notification or a command needn't be written:

services.AddMessageBus(bus => bus
    .WithInputQueue("discordmanagement")
    .ForwardEvent<RankChangedEventMessage>(message => new RankChangedEvent(message))          // a notification
    .ForwardEvent<ShardDeliveryDisabledEventMessage>(message => new DisableDeliveryCommand(message.ShardId)));  // a command

Failures, without exceptions. A handler (or a forwarded command) returns a Result, and the bus reads its errors:

An exception is a bug, not an answer, but it's caught and handled like the last row; only a shutdown leaves the event for the next instance. A retry starts at the handler that failed, so the ones before it don't run twice. The number of tries follows the bus' retry strategy (options.RetryStrategy(maxDeliveryAttempts: ...) in a configuration strategy).

• Delivery is at-least-once. A redelivered event, or one published again by a retried outbox message, can arrive twice. SkipDuplicateEvents() handles each one once: an event published from the outbox gets a message ID derived from the outbox message (IOutboxMessageContext), the same on every retry, and the subscriber remembers the IDs it handled in its ICachingService (3 hours by default: an outbox retry or a redelivery comes within minutes; each entry expires then). Over an in-memory cache each instance remembers its own; over Redis, the whole service does. Without it, handling an event twice must be harmless. The ID is claimed before the handlers run, in one step (IClaimStore.TryClaimForAsync: the registered claim store, or the registered ICachingService when it is one, as DistributedCachingService is), and a copy that arrives while the event is being handled on the same instance is skipped without asking the cache, so two copies arriving together aren't both handled. Across instances that holds as far as the cache claims atomically: Redis with SET NX does, DistributedCachingService only within one process. The claim is released when the event goes to the error queue or the handling stops (a shutdown), so its next delivery is handled rather than skipped.
• HandleOneAtATime() handles an event type one at a time on each instance (MassTransit's ConcurrentMessageLimit = 1).
• A subscription is a binding on the broker and outlives the service. When you delete a handler, keep RemoveSubscription<TEvent>() (or RemoveSubscription("TEvent") once the class is gone) for a release: it unbinds at start-up. Until then its events fail ("no message type here is called …"), count as messagebus.messages.unknown, and end up in the error queue.

The error queue: show, delete, move back

The events whose handlers kept failing wait in the service's <input queue>.error. IErrorQueue (registered by AddMessageBus, its contract in Resrcify.SharedKernel.Abstractions.MessageBus) shows them, deletes them, or moves them back to the queue they failed on, to be handled again once the cause is fixed:

var listed = await errorQueue.ListAsync(max: 100);          // Result<ErrorQueueContents>: the queue, its count, the first messages
// each FailedMessage: MessageId, MessageType (wire name), SourceQueue, Error (the handler and its errors),
// SentOnUtc, Headers, Text (a JSON body as text, unzipped) and Body
await errorQueue.RetryAsync(listed.Value.Messages[0].MessageId);   // moves it back
await errorQueue.RetryAllAsync("ShardRankChanged");                // every one of a type (or every one: null)
await errorQueue.DeleteAsync(messageId);
await errorQueue.DeleteAllAsync();

The Resrcify.SharedKernel.Web package maps it to admin endpoints (app.MapErrorQueueAdministration(), under /admin/error-queue), next to the outbox's.

• A message moved back resumes where it failed. It keeps its message ID and loses its error details. The handlers that had finished with it (handled it, or rejected it as the event's fault) before the one that failed are named in its resrcify-handled-by header and don't run again; the one that failed and those after it do. A service that skips duplicates handles it: its claim was released when it went to the error queue. If it fails again, it goes back to the error queue, with the new error.
• Results, not exceptions: ErrorQueue.MessageNotFound (NotFound) for an unknown ID, ErrorQueue.NotMoved (Conflict) for a message that can't be moved back (its queue is gone; it stays), ErrorQueue.None (NotFound) for a service without an input queue, ErrorQueue.Unavailable (ExternalFailure) when the broker can't be reached. Every deletion and move is logged.
• Listing uses nothing up. Each operation reads the queue once (basic.get), holding the messages until it is done, and hands back those it keeps with a requeueing basic.nack, which doesn't count as a delivery, and waits until a quorum queue has them back. A message moved back is published (mandatory, confirmed) before it leaves the error queue.
• It works on the in-memory network too (UseInMemory, the test harness).

Running on a RabbitMQ cluster

With more than one node, make the queues quorum queues and lift their delivery limit, on the broker: quorum as the vhost's default queue type, and a policy "delivery-limit": -1 for quorum queues. Classic queues aren't replicated, and a quorum queue otherwise drops a message the bus handed back 20 times (a service stopping, a health gate closing). The commands, and how to move existing queues: docs/OPERATIONS.md. Point the services at the cluster's one address (in Kubernetes, its Service), not at a list of nodes.

Running without RabbitMQ: in memory

UseInMemory() runs every bus of the service in memory: scatter-gather, rate-limited queues and publish/subscribe all work within the process. Use it for local runs without a broker and in tests; messages don't survive a restart.

services.AddMessageBus(bus =>
{
    if (options.Host is { Length: > 0 })
        bus.UseRabbitMq(new RabbitMqConnection(options.Host, options.Port, options.Username, options.Password));
    else
        bus.UseInMemory();
    ...
});

To test several services together, give their hosts the same network: var network = new InMemNetwork(); then bus.UseInMemory(network) in each.

For tests, the Resrcify.SharedKernel.MessageBus.Testing package does this for a service configured for RabbitMQ (services.AddMessageBusTestHarness(network)), and records what every bus published, sent, consumed and failed, waits until all are idle, injects failures and stands in for other services' queues. See its README.

Checking that two copies of a message agree

Each service keeps its own class for a message it shares, so nothing notices when one copy drifts (a renamed property, a changed nested class). One line in a test does:

[Fact]
public void SentinelArenaUnitsChanged_ShouldReadAsDiscordsCopy()
    => MessageContract.Differences<Sentinel.SentinelArenaUnitsChanged, Discord.SentinelArenaUnitsChanged>(Sample)
        .ShouldBeEmpty();

Differences writes the sample as the sender's bus does (pass the sender's serialization strategy, if it has one), reads it as the receiver's class and lists every property dropped, read differently, or not sent. Give the sample a value in every property, nested ones and lists included.

Where the sender's class isn't at hand (another repository), check the real message as it arrived instead: in a test handler on the receiving side, read the body of the message being handled and pass it with this side's class, MessageContract.Differences<Discord.SentinelArenaUnitsChanged>(json) (or Differences(type, json)). An end-to-end test that runs the real sender then fails on a renamed or missing property, whether or not it asserts that field.

Large messages: compression

Message bodies of 32 KB or more are gzipped on every bus (CompressMessagesAbove(bytes) changes the threshold, null turns it off). Every bus reads a gzipped message, whatever its own setting. Rebus.Diagnostics' message-size metric shows what is actually sent.

The format is Rebus' own EnableCompression (the body gzipped, header rbs2-content-encoding: gzip, same threshold rule), so a service on plain Rebus compression reads what this bus sends and the other way round. It is faster: it compresses at CompressionLevel.Fastest (Rebus: Optimal), and unzips into a buffer sized from the gzip trailer instead of a growing stream (falling back to growing when the trailer can't be believed). The compressed body is larger. For a 755 KB roster message (MessageCompressionBenchmarks in the PerformanceTests): zipping takes 1.8 ms instead of 7.2 ms, unzipping 0.86 ms and 0.74 MB allocated instead of 0.82 ms and 2.7 MB, and the body is 313 KB instead of 189 KB. A forwarded or dead-lettered message stays compressed.

Customising: strategies

Like the outbox's IOutboxInsertStrategy, the bus is extended through strategy interfaces (Resrcify.SharedKernel.MessageBus.Abstractions) rather than by subclassing:

internal sealed class RetryAndConfirmsStrategy : IBusConfigurationStrategy
{
    public void ConfigureTransport(RabbitMqOptionsBuilder transport)
        => transport.SetPublisherConfirms(true);

    public void ConfigureOptions(OptionsConfigurer options)
        => options.RetryStrategy(maxDeliveryAttempts: 10);
}

services.AddMessageBus(bus => bus
    .UseRabbitMq(...)
    .UseConfigurationStrategy(new RetryAndConfirmsStrategy())   // or a factory: provider => ...
    .AddRateLimitedQueue<GuildRequest, GuildResponse, GuildHandler>(
        queue => queue.RateLimiter = new SlidingWindowStrategy()));

A limiter is created each time a queue's bus starts (including after a health-gate restart) and disposed when it stops.

Health check

services.AddHealthChecks().AddMessageBus(tags: ["ready"]);

(The queues' own gate is separate: every queue on a health-check tag shares one check of it, every HealthCheckInterval (the shortest any of them asks for) and early after a failed request, so an upstream's check runs once per instance, not once per queue.)

Unhealthy (or the failureStatus you pass) while RabbitMQ is unreachable; always healthy in memory. Its data lists each rate-limited queue as consuming or stepped aside. A queue that stepped aside doesn't change the status: its own health check (e.g. the upstream's) already says why. Never give it the tag a rate-limited queue is gated on: the queue would stop on every broker blip, which it already recovers from by itself.

Tracing and metrics

Every bus emits OpenTelemetry-compatible spans (Rebus.Diagnostics): a span per send and per handled message, with the trace context carried in the message headers, so a responder's handling joins the requester's trace. Rebus only traces a send that has a parent span, which the outbox provides: every processed outbox message gets a span (OutboxDiagnostics, in UnitOfWork). One scatter-gather batch is then one trace, with each request's send, its handling in the provider, and the reply.

services.AddOpenTelemetry()
    .WithTracing(tracing => tracing
        .AddSource(OutboxDiagnostics.ActivitySourceName)            // Resrcify.SharedKernel.UnitOfWork.Outbox
        .AddSource(RebusDiagnosticConstants.ActivitySourceName))    // Rebus.Diagnostics
    .WithMetrics(metrics => metrics
        .AddMeter(RebusDiagnosticConstants.MeterName)               // message counts, sizes and delays
        .AddMeter(MessageBusDiagnostics.MeterName));                // Resrcify.SharedKernel.MessageBus.Diagnostics

Delivery is at-least-once, so a request can be answered more than once (e.g. when a responder's bus stops after it answered but before it acknowledged). The requester takes the first answer per item and ignores the rest; messagebus.scatter_gather.replies (by outcome: accepted, duplicate, late) shows how many there are.

Rebus.OpenTelemetry's AddRebusInstrumentation() is the same AddSource call, so it isn't needed.

The package's own metrics (MessageBusDiagnostics.MeterName):

Behaviour you can rely on