Policy-Based Routing¶

Verified by tests

PolicyEngineTests, PolicyContextTests, PolicyConfigurationExtensionsTests, PolicyConfigurationTransportTests, PolicyContextPoolTests, PolicyDecisionTests, PolicyDecisionTrailTests — library CI run #31657041675 (2026-08-13)

Policy-based routing enables dynamic message configuration based on runtime conditions. Policies evaluate message context (type, aggregate ID, tenant, environment) and return routing configuration (topics, execution strategies, partitioning) without hardcoding business logic into handlers.

Why Policies?¶

Policies decouple routing decisions from business logic:

Without Policies	With Policies	Benefit
Hardcoded Routes	Dynamic predicates	Flexible configuration
If/Else Chains	First-match evaluation	Clean code
Per-Handler Config	Centralized policy engine	Single source of truth
No Audit Trail	PolicyDecisionTrail	Full observability
Multi-Tenant Logic Scattered	Tenant-based policies	Centralized multi-tenancy

Use Cases: - ✅ Multi-Tenancy - Route messages to tenant-specific topics/databases - ✅ Environment-Based Routing - Different config for dev/staging/prod - ✅ Aggregate-Based Routing - Route by message aggregate type - ✅ Execution Strategies - Serial vs parallel based on message type - ✅ Feature Flags - Enable/disable routing based on tags/metadata

Architecture¶

Policy Evaluation Flow¶

flowchart TD
    subgraph Processing["Message Processing"]
        Rent["1. Rent PolicyContext from pool<br/>context = PolicyContextPool.Rent(message, ...)"]
        Evaluate["2. Evaluate policies<br/>config = await policyEngine.MatchAsync(context)"]
        Use["3. Use configuration<br/>- Topic routing<br/>- Execution strategy<br/>- Partitioning<br/>- Concurrency"]

        Rent --> Evaluate --> Use
    end

PolicyEngine Evaluation:

flowchart TD
    Match["PolicyEngine.MatchAsync(context)<br/>Policies evaluated in order (first match wins)"]
    Policy1["Policy 1: TenantRouting<br/>Predicate: context.GetMetadata(&quot;tenantId&quot;) == &quot;tenant-a&quot;<br/>Matched: ✅<br/>Configuration: Topic = &quot;tenant-a-events&quot;"]
    Return["⭐ RETURN (first match - skip remaining policies)"]
    Policy2["Policy 2: EnvironmentRouting (skipped)"]
    Policy3["Policy 3: DefaultRouting (skipped)"]

    Match --> Policy1
    Policy1 --> Return
    Match -.-> Policy2
    Match -.-> Policy3

PolicyDecisionTrail (Observability) - context.Trail.Decisions:

Index	PolicyName	Rule	Matched	Reason
[0]	"TenantRouting"	"predicate"	✅	"Policy predicate matched"
[1]	"EnvironmentRouting"	"predicate"	❌	"Policy predicate did not match"

Core Components¶

1. PolicyEngine¶

Purpose: Evaluates policies in order, returns the first match.

Registration — the engine is a plain class with a parameterless constructor; register it (typically as a singleton) and add policies imperatively:

Register PolicyEngine and add policies

var policyEngine = new PolicyEngine();

// Add policies (evaluated in the order they are added)
policyEngine.AddPolicy(
  name: "TenantRouting",
  predicate: context =>
    context.GetMetadata("tenantId")?.ToString() == "tenant-a",
  configure: config =>
    config.PublishToServiceBus("tenant-a-events")
);

policyEngine.AddPolicy(
  name: "DefaultRouting",
  predicate: context => true,  // Always matches (fallback)
  configure: config =>
    config.PublishToServiceBus("default-events")
);

// Evaluate policies against a context
var config = await policyEngine.MatchAsync(context);

AddPolicy(string name, Func<PolicyContext, bool> predicate, Action<PolicyConfiguration> configure): - name is required — AddPolicy throws ArgumentException if it is null, empty, or whitespace. - predicate and configure are required — AddPolicy throws ArgumentNullException if either is null.

MatchAsync(PolicyContext context) evaluation rules: - Policies are evaluated in registration order. - The first matched policy's configure delegate runs against a fresh PolicyConfiguration, which is returned. - Subsequent policies are skipped once a match is found. - If no policy matches, MatchAsync returns null. - MatchAsync throws ArgumentNullException if context is null.

2. PolicyContext¶

Purpose: Universal context with message, envelope, services, environment, and the decision trail.

Properties: PolicyContext properties

public class PolicyContext {
  public object Message { get; }               // The message being processed
  public Type MessageType { get; }             // Runtime type of the message
  public IMessageEnvelope? Envelope { get; }   // Envelope with metadata (may be null)
  public IServiceProvider? Services { get; }   // DI container (may be null)
  public string Environment { get; }           // Default: "development" (lowercase)
  public DateTimeOffset ExecutionTime { get; } // When the context was created (UTC)
  public PolicyDecisionTrail Trail { get; }    // Decision audit trail
}

The public constructor is PolicyContext(object message, IMessageEnvelope? envelope = null, IServiceProvider? services = null, string environment = "development"). The first positional argument is the message (not the envelope), and Environment defaults to the lowercase string "development" — match on that exact casing in predicates.

Helper Methods: PolicyContext helper methods

// Service resolution (throws if Services is null or the service isn't registered)
var repository = context.GetService<IOrderRepository>();

// Metadata access (reads from the envelope; null-safe)
var tenantId = context.GetMetadata("tenantId");

// Tags — read from the "tags" metadata key (string array / JSON array)
var hasHighPriority = context.HasTag("high-priority");

// Flags — read bitwise from the "flags" metadata key; accepts any [Flags] enum
var isUrgent = context.HasFlag(ProcessingOptions.Urgent);

// Aggregate matching — true when the message type name contains the aggregate name
bool isOrderMessage = context.MatchesAggregate<Order>();

// Aggregate/stream ID extraction (zero reflection).
// Requires a [StreamId] Guid property on the message AND a registered
// IStreamIdExtractor (call services.AddWhizbang() at startup); throws otherwise.
var orderId = context.GetAggregateId();

Notes on the helpers: - GetService<T>() is constrained to where T : class. It throws InvalidOperationException when Services is null, and again when the requested service is not registered. - HasFlag takes any Enum (typically a user-defined [Flags] enum) and reads the "flags" metadata value as a bitwise mask. - HasTag matches a string against the "tags" metadata value (supports a JsonElement array, string[], or IEnumerable<string>). - MatchesAggregate<T>() uses a naming convention — it returns true when the message type name contains the aggregate type name (case-insensitive), e.g. CreateOrder matches Order. - GetAggregateId() returns a Guid and requires all of: a configured IServiceProvider, a registered IStreamIdExtractor, and a [StreamId]-marked Guid property on the message.

Pooling — PolicyContext is designed to be reused to minimize allocations. Rent from the static pool and always return it: Rent and return a pooled PolicyContext

// Rent from the pool (creates a new instance only when the pool is empty)
var context = PolicyContextPool.Rent(message, envelope, services, "production");

try {
  var config = await policyEngine.MatchAsync(context);
  // Use config...
} finally {
  // Always return to the pool (safe to call with null)
  PolicyContextPool.Return(context);
}

3. PolicyDecisionTrail¶

Purpose: Records every policy decision for debugging and time-travel.

Usage: Inspect the policy decision trail

// PolicyEngine records a decision for every policy it evaluates.
var config = await policyEngine.MatchAsync(context);

// Query the trail
var matchedPolicies = context.Trail.GetMatchedRules();
var unmatchedPolicies = context.Trail.GetUnmatchedRules();

foreach (var decision in context.Trail.Decisions) {
  Console.WriteLine($"{decision.PolicyName}: {decision.Matched} - {decision.Reason}");
}

Each PolicyDecision carries PolicyName, Rule, Matched, Configuration, Reason, and Timestamp. The engine records Rule = "predicate" with Reason = "Policy predicate matched" / "Policy predicate did not match", and "Evaluation failed: {message}" when a predicate throws.

Benefits: - Debugging: See why a specific configuration was applied. - Auditing: Track policy decisions over time. - Time-Travel: Replay message processing with decision history.

4. PolicyConfiguration¶

Purpose: Routing and execution configuration returned by the matched policy.

Properties: PolicyConfiguration properties

public class PolicyConfiguration {
  // Publishing (outbound) and subscribing (inbound) transport targets
  public List<PublishTarget> PublishTargets { get; }
  public List<SubscriptionTarget> SubscriptionTargets { get; }

  // Routing
  public string? Topic { get; }
  public string? StreamId { get; }

  // Execution / partitioning
  public Type? ExecutionStrategyType { get; }
  public Type? PartitionRouterType { get; }
  public Type? SequenceProviderType { get; }
  public int? PartitionCount { get; }
  public int? MaxConcurrency { get; }

  // Persistence size limits (JSONB columns)
  public int? MaxDataSizeBytes { get; }
  public bool SuppressSizeWarnings { get; }
  public bool ThrowOnSizeExceeded { get; }
}

Fluent API — the configure delegate receives a PolicyConfiguration (not the context), so every setter takes a constant value. Per-message stream derivation (e.g. building a stream key from an aggregate ID at evaluation time) is not available inside configure today; UseStreamId takes a fixed string.

Configure routing, execution, and persistence

configure: config => config
  .UseTopic("order-events")
  .UseStreamId("order-stream")
  .UseExecutionStrategy<SerialExecutor>()
  .UsePartitionRouter<HashPartitionRouter>()
  .UseSequenceProvider<InMemorySequenceProvider>()
  .WithPartitions(count: 100)
  .WithConcurrency(maxConcurrency: 10)
  .WithPersistenceSize(maxDataSizeBytes: 7000, throwOnExceeded: true)

UseTopic(string) sets the logical Topic; UseStreamId(string) sets the StreamId (ordering/partitioning key).
UseExecutionStrategy<T>() — the shipped strategies are SerialExecutor (strict FIFO, one message at a time) and ParallelExecutor (concurrent, bounded by WithConcurrency).
UsePartitionRouter<T>() — HashPartitionRouter (consistent hashing over the stream key) ships in Whizbang.Core.Partitioning.
UseSequenceProvider<T>() — InMemorySequenceProvider (monotonic per-stream sequence numbers) ships in Whizbang.Core.Sequencing.
WithPartitions(int), WithConcurrency(int), and WithPersistenceSize(maxDataSizeBytes: …) each throw ArgumentOutOfRangeException when the supplied value is <= 0.

Transport targets — beyond the logical Topic, PolicyConfiguration can add concrete publish and subscribe targets per transport:

Add transport publish and subscribe targets

configure: config => config
  // Publishing (outbound)
  .PublishToKafka("orders")
  .PublishToServiceBus("order-events")
  .PublishToRabbitMQ("orders-exchange", routingKey: "orders.created")
  // Subscribing (inbound)
  .SubscribeFromKafka("orders", consumerGroup: "order-workers")
  .SubscribeFromServiceBus("order-events", subscriptionName: "order-sub")
  .SubscribeFromRabbitMQ("orders-exchange", queueName: "orders-queue");

Common Policies¶

1. Multi-Tenant Routing¶

Multi-tenant routing by tenant metadata

policyEngine.AddPolicy(
  name: "TenantARouting",
  predicate: context =>
    context.GetMetadata("tenantId")?.ToString() == "tenant-a",
  configure: config => config
    .PublishToServiceBus("tenant-a-events")
    .UseStreamId("tenant-a-orders")
);

policyEngine.AddPolicy(
  name: "TenantBRouting",
  predicate: context =>
    context.GetMetadata("tenantId")?.ToString() == "tenant-b",
  configure: config => config
    .PublishToServiceBus("tenant-b-events")
    .UseStreamId("tenant-b-orders")
);

// Fallback for unknown tenants
policyEngine.AddPolicy(
  name: "DefaultTenantRouting",
  predicate: context => true,
  configure: config => config
    .PublishToServiceBus("default-events")
);

2. Environment-Based Routing¶

Match on context.Environment, which defaults to the lowercase "development":

Environment-based routing

policyEngine.AddPolicy(
  name: "ProductionRouting",
  predicate: context => context.Environment == "production",
  configure: config => config
    .PublishToServiceBus("prod-events")
    .WithConcurrency(maxConcurrency: 50)
);

policyEngine.AddPolicy(
  name: "StagingRouting",
  predicate: context => context.Environment == "staging",
  configure: config => config
    .PublishToServiceBus("staging-events")
    .WithConcurrency(maxConcurrency: 10)
);

policyEngine.AddPolicy(
  name: "DevelopmentRouting",
  predicate: context => context.Environment == "development",
  configure: config => config
    .PublishToServiceBus("dev-events")
    .WithConcurrency(maxConcurrency: 1)  // Serial processing in dev
);

3. Aggregate-Based Routing¶

Match on the aggregate type via the naming convention, then set a stream id and partitioning:

Aggregate-based routing and partitioning

policyEngine.AddPolicy(
  name: "OrderPartitioning",
  predicate: context => context.MatchesAggregate<Order>(),
  configure: config => config
    .UseStreamId("orders")
    .UsePartitionRouter<HashPartitionRouter>()
    .WithPartitions(count: 100)
);

policyEngine.AddPolicy(
  name: "CustomerPartitioning",
  predicate: context => context.MatchesAggregate<Customer>(),
  configure: config => config
    .UseStreamId("customers")
    .UsePartitionRouter<HashPartitionRouter>()
    .WithPartitions(count: 50)
);

4. Message Type-Based Execution¶

Execution strategy by message type

policyEngine.AddPolicy(
  name: "BulkImportExecutionStrategy",
  predicate: context => context.MessageType.Name.Contains("BulkImport"),
  configure: config => config
    .UseExecutionStrategy<ParallelExecutor>()
    .WithConcurrency(maxConcurrency: 100)
);

policyEngine.AddPolicy(
  name: "OrderExecutionStrategy",
  predicate: context => context.MessageType.Name.Contains("Order"),
  configure: config => config
    .UseExecutionStrategy<SerialExecutor>()  // Strict ordering for orders
);

5. Tag-Based Routing¶

Tag-based routing

policyEngine.AddPolicy(
  name: "HighPriorityRouting",
  predicate: context => context.HasTag("high-priority"),
  configure: config => config
    .PublishToServiceBus("priority-events")
    .WithConcurrency(maxConcurrency: 100)
);

policyEngine.AddPolicy(
  name: "ArchivalRouting",
  predicate: context => context.HasTag("archival"),
  configure: config => config
    .PublishToServiceBus("archive-events")
    .WithConcurrency(maxConcurrency: 1)  // Low priority
);

Advanced Patterns¶

Composite Policies¶

Composite predicate combining conditions

policyEngine.AddPolicy(
  name: "HighValueOrderRouting",
  predicate: context => {
    bool isOrder = context.MatchesAggregate<Order>();
    bool isHighValue = context.GetMetadata("totalAmount") is decimal amount && amount > 10000;
    bool isProduction = context.Environment == "production";

    return isOrder && isHighValue && isProduction;
  },
  configure: config => config
    .PublishToServiceBus("high-value-orders")
    .UseExecutionStrategy<SerialExecutor>()
    .WithConcurrency(maxConcurrency: 1)
);

Service-Injected Policies¶

Resolve a service inside a predicate

policyEngine.AddPolicy(
  name: "FeatureFlagRouting",
  predicate: context => {
    // Resolve a service from the context's IServiceProvider
    var featureFlags = context.GetService<IFeatureFlagService>();
    return featureFlags.IsEnabled("new-event-routing");
  },
  configure: config => config
    .PublishToServiceBus("new-events-topic")
);

Time-Based Policies¶

Time-based routing on ExecutionTime

policyEngine.AddPolicy(
  name: "PeakHoursRouting",
  predicate: context => {
    var hour = context.ExecutionTime.Hour;
    return hour >= 9 && hour <= 17;  // 9 AM - 5 PM
  },
  configure: config => config
    .WithConcurrency(maxConcurrency: 100)  // High concurrency during peak
);

policyEngine.AddPolicy(
  name: "OffHoursRouting",
  predicate: context => true,  // Fallback
  configure: config => config
    .WithConcurrency(maxConcurrency: 10)  // Lower concurrency off-peak
);

Testing Policies¶

Unit Testing Predicates¶

Unit test a tenant predicate and its publish target

[Test]
public async Task TenantARouting_WithTenantA_MatchesAsync() {
  // Arrange
  var context = new PolicyContext(
    message: new CreateOrder(),
    envelope: CreateEnvelope(metadata: new Dictionary<string, object> {
      ["tenantId"] = "tenant-a"
    }),
    services: null,
    environment: "production"
  );

  var policyEngine = new PolicyEngine();
  policyEngine.AddPolicy(
    name: "TenantARouting",
    predicate: ctx => ctx.GetMetadata("tenantId")?.ToString() == "tenant-a",
    configure: config => config.PublishToServiceBus("tenant-a-events")
  );

  // Act
  var result = await policyEngine.MatchAsync(context);

  // Assert
  await Assert.That(result).IsNotNull();
  await Assert.That(result!.PublishTargets).HasCount().EqualTo(1);
  await Assert.That(result.PublishTargets[0].Destination).IsEqualTo("tenant-a-events");
}

Testing Topic and Stream Id¶

Assert Topic and StreamId on the matched configuration

[Test]
public async Task OrderPolicy_SetsTopicAndStreamIdAsync() {
  // Arrange
  var context = new PolicyContext(new CreateOrder(), null, null, "production");

  var policyEngine = new PolicyEngine();
  policyEngine.AddPolicy(
    name: "OrderPolicy",
    predicate: ctx => true,
    configure: config => config
      .UseTopic("orders")
      .UseStreamId("order-123")
  );

  // Act
  var config = await policyEngine.MatchAsync(context);

  // Assert
  await Assert.That(config).IsNotNull();
  await Assert.That(config!.Topic).IsEqualTo("orders");
  await Assert.That(config.StreamId).IsEqualTo("order-123");
}

Testing Policy Order¶

Verify first-match-wins and skip

[Test]
public async Task PolicyEngine_FirstMatchWins_SkipsSubsequentPoliciesAsync() {
  // Arrange
  var context = new PolicyContext(new CreateOrder(), null, null, "production");

  var policyEngine = new PolicyEngine();

  policyEngine.AddPolicy("FirstPolicy",
    predicate: ctx => true,  // Always matches
    configure: config => config.PublishToServiceBus("first-topic")
  );

  policyEngine.AddPolicy("SecondPolicy",
    predicate: ctx => true,  // Would match, but skipped
    configure: config => config.PublishToServiceBus("second-topic")
  );

  // Act
  var result = await policyEngine.MatchAsync(context);

  // Assert
  await Assert.That(result!.PublishTargets[0].Destination).IsEqualTo("first-topic");

  // Verify decision trail
  var matched = context.Trail.GetMatchedRules().ToList();
  await Assert.That(matched).HasCount().EqualTo(1);
  await Assert.That(matched[0].PolicyName).IsEqualTo("FirstPolicy");
}

Testing PolicyDecisionTrail¶

Assert every policy is recorded in the trail

[Test]
public async Task PolicyEngine_RecordsDecisionTrail_ForAllPoliciesAsync() {
  // Arrange
  var context = new PolicyContext(new CreateOrder(), null, null, "production");

  var policyEngine = new PolicyEngine();
  policyEngine.AddPolicy("Policy1", ctx => false, config => { });
  policyEngine.AddPolicy("Policy2", ctx => true, config => { });

  // Act
  await policyEngine.MatchAsync(context);

  // Assert
  var decisions = context.Trail.Decisions.ToList();
  await Assert.That(decisions).HasCount().EqualTo(2);

  // First policy did not match
  await Assert.That(decisions[0].PolicyName).IsEqualTo("Policy1");
  await Assert.That(decisions[0].Matched).IsFalse();

  // Second policy matched
  await Assert.That(decisions[1].PolicyName).IsEqualTo("Policy2");
  await Assert.That(decisions[1].Matched).IsTrue();
}

Best Practices¶

DO ✅¶

✅ Register policies in order of specificity (most specific first, fallback last)
✅ Give every policy a non-empty name (AddPolicy throws on null/empty/whitespace names)
✅ Return contexts to the pool after policy evaluation
✅ Test policies with various inputs (unit test predicates)
✅ Use service injection for complex predicates (feature flags, config)
✅ Add a fallback policy with predicate: ctx => true at the end
✅ Monitor PolicyDecisionTrail in logs for debugging

DON'T ❌¶

❌ Perform expensive operations in predicates (database queries, API calls)
❌ Mutate context in predicates (side effects)
❌ Rely on a predicate exception to route — a throwing predicate is caught, recorded as a failed decision, and the engine continues to the next policy
❌ Skip returning contexts to the pool (extra allocations)
❌ Hardcode business logic in predicates (use services instead)
❌ Expect per-message stream derivation inside configure (it only sees a PolicyConfiguration, so UseStreamId is a constant string)

Troubleshooting¶

Problem: No Policy Matches, Null Configuration¶

Symptoms: MatchAsync() returns null, no configuration applied.

Cause: No policies registered or all predicates return false.

Solution: Add a fallback policy and inspect the trail

// Add fallback policy
policyEngine.AddPolicy(
  name: "DefaultPolicy",
  predicate: context => true,  // Always matches (last resort)
  configure: config => config
    .PublishToServiceBus("default-events")
);

// Verify policies registered
var config = await policyEngine.MatchAsync(context);
if (config is null) {
  logger.LogWarning("No policy matched for message {MessageType}", context.MessageType.Name);

  // Check decision trail
  foreach (var decision in context.Trail.Decisions) {
    logger.LogDebug("Policy {PolicyName}: {Matched} - {Reason}",
      decision.PolicyName, decision.Matched, decision.Reason);
  }
}

Problem: Wrong Policy Matched¶

Symptoms: Unexpected configuration returned.

Cause: Policy order incorrect (fallback registered before specific policies).

Solution: Order specific policies before the fallback

// ❌ WRONG: Fallback first (always matches)
policyEngine.AddPolicy("Fallback", ctx => true, config => config.PublishToServiceBus("default"));
policyEngine.AddPolicy("Specific", ctx => ctx.HasTag("high-priority"), config => config.PublishToServiceBus("priority"));

// ✅ CORRECT: Specific first, fallback last
policyEngine.AddPolicy("Specific", ctx => ctx.HasTag("high-priority"), config => config.PublishToServiceBus("priority"));
policyEngine.AddPolicy("Fallback", ctx => true, config => config.PublishToServiceBus("default"));

Problem: Predicate Throws Exception¶

Symptoms: Policy skipped with an error recorded in the decision trail.

Cause: An exception thrown in the predicate. The engine catches it, records a failed decision (Reason: "Evaluation failed: …"), and continues to the next policy.

Solution: Make predicates null-safe

// Predicate that can throw — the engine catches it and records the failure
policyEngine.AddPolicy(
  name: "FaultyPolicy",
  predicate: context => {
    var metadata = context.GetMetadata("value");
    return (int)metadata! > 100;  // NullReferenceException if missing
  },
  configure: config => { }
);

// Decision trail shows:
//   PolicyName: "FaultyPolicy"
//   Matched: false
//   Reason: "Evaluation failed: Object reference not set to an instance of an object."

// FIX: null-safe predicate
policyEngine.AddPolicy(
  name: "SafePolicy",
  predicate: context => {
    var metadata = context.GetMetadata("value");
    return metadata is int value && value > 100;  // ✅ Null-safe
  },
  configure: config => { }
);