Cirreum.Runtime.Messaging 2.0.0

Cirreum.Runtime.Messaging

High-performance distributed messaging with policy-driven batching and observability for .NET applications

Overview

Cirreum.Runtime.Messaging composes the messaging stack for Cirreum server hosts and implements the Cirreum Distributed Messaging feature. A single AddMessaging() call does both:

• Messaging services composition — registers the configured messaging providers (Azure Service Bus) from Cirreum:Messaging:Providers, exposing each instance as a keyed IMessagingClient (queues, topics, subscriptions) with health checks and tracing. Apps can consume these clients directly for their own messaging workflows, with or without the distributed-messaging layer.
• Distributed Messaging — the runtime delivery engine for the DistributedMessage channel. The messaging model — DistributedMessage, the wire envelope, the registry, and the IBatchingPolicy strategy — ships in Cirreum.Messaging.Distributed; this package provides the moving parts: the outbound Conductor bridge, the policy-driven batching processor, the transport publisher, the inbound receiver, and OpenTelemetry metrics, offering both synchronous and batched delivery with built-in resilience.

Key Features

🚀 Flexible Message Delivery

• Dual-mode publishing: Direct (synchronous) and background (batched) delivery, per message or by channel default
• Transport abstraction: Pluggable providers (Azure Service Bus included)
• Message targeting: Support for both queue-based events and topic-based notifications via [DistributedMessageTarget]

📦 Policy-Driven Batching System

• Pluggable batch sizing: each batch's capacity and fill window come from the channel's IBatchingPolicy, fed live queue-depth, send-rate, and error-rate observables; the default policy passes the configured base values through
• Priority queuing: High-priority messages with rate limiting and automatic age promotion
• Circuit breaker: Built-in fault tolerance for resilient message delivery

📊 Comprehensive Observability

• OpenTelemetry integration: Distributed tracing and metrics collection
• Lifecycle tracking: Monitor messages from receipt to delivery
• Queue depth alerts: Configurable thresholds for proactive monitoring
• Performance metrics: Detailed timing for queue and delivery operations

⚙️ Production-Ready

• Thread-safe operations: Designed for high-concurrency scenarios
• Graceful shutdown: Proper cleanup of background services
• Health checks: Transport-instance health monitoring (queue/topic validation, readiness participation) via the provider configuration
• Structured logging: Rich context for troubleshooting

📥 Inbound Message Dispatch (added 1.1.0)

• Hosted receiver: DistributedMessageReceiver consumes from queue and/or topic subscription concurrently
• Conductor dispatch: Handlers are standard INotificationHandler<DistributedMessageReceived<T>> — auto-discovered, scoped, pipeline-aware
• Self-echo prevention: cirreum.node application property + replica identity (INodeIdProvider) skip own publishes pre-deserialization
• Cross-broker filterable metadata: Four application properties (cirreum.identifier, cirreum.version, cirreum.producer, cirreum.node) stamped on every outbound message for broker-side subscription filtering
• Multi-head ready: Per-deployment SubscriptionName differentiates heads; same binary, different config; broker fan-outs messages to all heads

Quick Start

Installation

dotnet add package Cirreum.Runtime.Messaging

Basic Setup

var builder = DomainApplication.CreateBuilder(args);

// Registers the Service Bus provider, the delivery engine, the outbound
// Conductor bridge, and (when configured) the inbound receiver.
builder.AddMessaging();

var app = builder.Build();
await app.RunAsync();

Direct IMessagingClient Usage

AddMessaging() registers every instance under Cirreum:Messaging:Providers as a keyed IMessagingClient — usable directly for app-owned queues and topics, with or without the distributed-messaging layer:

public sealed class InvoiceQueueService(
	[FromKeyedServices("default")] IMessagingClient client) {

	public Task EnqueueAsync(Invoice invoice, CancellationToken ct) =>
		client.UseQueueSender("invoices.pending.v1")
			.PublishMessageAsync(OutboundMessage.AsJsonContent(invoice), ct);
}

The client surface (queue senders/receivers, topics, subscriptions, peek/defer/dead-letter) is defined by Cirreum.Messaging and implemented per broker by the provider package (e.g., Cirreum.Messaging.Azure) — see those packages for the full client reference. The Choosing a Dispatch Path section below covers when to use the client directly versus the distributed channel.

Defining and Publishing Messages

Messages derive from DistributedMessage, declare their wire contract with [MessageVersion], and optionally pick a routing target (topic is the default):

[MessageVersion("orders.created", "1.0")]
[DistributedMessageTarget(MessageTarget.Queue)]
public sealed record OrderCreatedEvent(string OrderId) : DistributedMessage;

Publish through Conductor — the outbound bridge forwards any published DistributedMessage to the configured transport automatically:

public sealed class OrderService(IPublisher publisher) {

	public async Task ProcessOrderAsync(Order order) {

		// Delivered per the channel default (direct or batched)
		await publisher.PublishAsync(new OrderCreatedEvent(order.Id));

		// Opt a specific message into batched background delivery
		await publisher.PublishAsync(new OrderCreatedEvent(order.Id) {
			UseBackgroundDelivery = true,
			Priority = DistributedMessagePriority.TimeSensitive
		});
	}
}

Configuration

{
  "Cirreum": {
	"Messaging": {
	  "Distributed": {
		"InstanceKey": "app-primary",
		"QueueName": "app-events",
		"TopicName": "app-notifications",
		"BackgroundDelivery": {
		  "UseBackgroundDeliveryByDefault": true,
		  "QueueCapacity": 1000,
		  "BatchCapacity": 10,
		  "BatchFillWaitTime": "00:00:00.0500000",
		  "CircuitBreakerThreshold": 5,
		  "CircuitResetTimeout": "00:01:00"
		},
		"Metrics": {
		  "QueueDepthWarningThreshold": 500,
		  "QueueDepthCriticalThreshold": 1000
		}
	  }
	}
  }
}

InstanceKey names the keyed IMessagingClient registration from the transport provider's Cirreum:Messaging:Providers configuration. See the Configuration Guide for every setting, defaults, and provider-instance configuration.

Custom Batching Policy

The default IBatchingPolicy passes the configured base values through unchanged. Apps that want dynamic batching plug in a policy via the composition callback — either the framework-supplied day-of-week / time-of-day scaler:

builder.AddMessaging(m => m.UseTimeOfDayBatching(schedule => {
	schedule.TimeZone = TimeZoneInfo.FindSystemTimeZoneById("Eastern Standard Time");
	schedule.Rules.Add(new() {
		Days = [DayOfWeek.Friday, DayOfWeek.Saturday, DayOfWeek.Sunday],
		StartHour = 16,
		EndHour = 23,
		ScalingFactor = 0.5, // high volume expected — halve the fill wait, send sooner
		Description = "Weekend evening spike"
	});
}));

or a fully custom policy (traffic-aware, queue-depth-aware, business-signal-aware):

builder.AddMessaging(m => m.UseBatchingPolicy<TrafficAwareBatchingPolicy>());

Each batch, the processor calls Evaluate(BatchingContext) with the configured base values plus live observables (current queue depth, rolling send rate, rolling error rate) and applies the returned fill wait time and capacity.

Consuming Inbound Messages (added 1.1.0)

Configure the receiver in appsettings:

{
  "Cirreum": {
	"Messaging": {
	  "Distributed": {
		"Receiver": {
		  "InstanceKey": "app-primary",
		  "TopicName": "app.notifications.v1",
		  "SubscriptionName": "api-head",
		  "MaxConcurrency": 1
		}
	  }
	}
  }
}

Implement handlers using the standard Conductor notification pattern — auto-discovered, no registration boilerplate.

The framework wraps every inbound message in DistributedMessageReceived<TMessage> which carries both the typed payload (Message) and the original wire envelope (Envelope) so handlers can inspect wire-level metadata without re-deserializing or threading additional context:

using Cirreum.Conductor;
using Cirreum.Messaging;
using Microsoft.Extensions.Logging;

public sealed class EvidenceInstanceChangeHandler(
	IEvidenceInstanceRegistry registry,
	ILogger<EvidenceInstanceChangeHandler> logger
) : INotificationHandler<DistributedMessageReceived<EvidenceInstanceChangedV1>> {
	public Task HandleAsync(
		DistributedMessageReceived<EvidenceInstanceChangedV1> notification,
		CancellationToken ct) {
		
		// The typed payload — strongly typed to the wrapped TMessage.
		var change = notification.Message;

		// The original wire envelope — wire-level metadata for audit, telemetry,
		// latency calculations, or replay detection.
		var envelope = notification.Envelope;

		logger.LogInformation(
			"Evidence instance {Key} changed (op={Operation}). "
			+ "From producer={Producer}, published={PublishedAt}, version={Version}.",
			change.Key,
			change.Operation,
			envelope.ProducerId,
			envelope.PublishedAt,
			envelope.MessageVersion);

		return registry.ApplyRemoteChangeAsync(change.Operation, change.Key, ct);
	}
}

The envelope properties available to every handler:

See the Configuration Guide for the receiver's full settings and queue-vs-subscription semantics.

Choosing a Dispatch Path

[MessageVersion] + DistributedMessage define a wire contract. Publishing through Conductor is one transport for that contract, not the only one — three patterns are valid, and serious business workflows often outgrow the first:

The framework path funnels everything through the channel's single configured queue/topic — deliberately, for framework-level eventing. When a workflow needs its own queue, keep Cirreum's envelope vocabulary (stable identifier + version, producer id, publish timestamp, resolvable type) and route it yourself (pattern 2):

var envelope = DistributedMessageEnvelope.Create(order, definition, producerId);
await messagingClient
	.UseQueueSender("orders.processing.v1")
	.PublishMessageAsync(OutboundMessage.AsJsonContent(envelope).WithSubject("orders.created.v1.0"));

Consumers on that queue can still re-materialize the payload with envelope.ResolveMessageType() / DeserializeMessage<T>(), and audit/observability tooling reads the same envelope shape across all three patterns.

Documentation

• Configuration Guide — every channel, background-delivery, receiver, metrics, and provider setting with defaults
• Migration Guide (v1 → v2) — breaking changes and the find/replace table for the 2.0 foundation-reset release
• Changelog

Contribution Guidelines

1. Be conservative with new abstractions
   The API surface must remain stable and meaningful.

2. Limit dependency expansion
   Only add foundational, version-stable dependencies.

3. Favor additive, non-breaking changes
   Breaking changes ripple through the entire ecosystem.

4. Include thorough unit tests
   All primitives and patterns should be independently testable.

5. Document architectural decisions
   Context and reasoning should be clear for future maintainers.

6. Follow .NET conventions
   Use established patterns from Microsoft.Extensions.* libraries.

Versioning

Cirreum.Runtime.Messaging follows Semantic Versioning:

• Major - Breaking API changes
• Minor - New features, backward compatible
• Patch - Bug fixes, backward compatible

License

This project is licensed under the MIT License - see the LICENSE file for details.

Cirreum Foundation Framework
Layered simplicity for modern .NET