Sweeper Circuit Breaking

Overview

Sweeper Circuit Breaking is a resilience feature that prevents failures to publish to one topic from blocking attempts to publish to other topics, when publishing messages from the Outbox. When a topic repeatedly fails to publish, the circuit breaker "trips" that topic, temporarily preventing further publish attempts until a cooldown period expires.

This feature is particularly valuable in scenarios where:

• Transport failures: A message broker or queue becomes unavailable

• Topic-specific issues: A specific topic/queue has configuration problems or capacity issues

• Cascade prevention: Failing topics would otherwise block the Outbox Sweeper from processing healthy topics

• Resource protection: Repeated failures to unhealthy topics consume resources without benefit

How It Work

The Sweeper Circuit Breaker operates at the topic level during Outbox clearing operations:

Normal Operation

1. Outbox Sweeper runs: Periodically attempts to clear outstanding messages from the Outbox

2. Messages grouped by topic: Messages are organized by their routing key (topic)

3. Publish attempts: The sweeper attempts to publish messages to their respective topics

4. Success: Messages are published and marked as dispatched

Circuit Breaking Behavior

When a topic fails to publish:

1. Failure detected: An exception occurs during message publication to a specific topic

2. Circuit trips: The circuit breaker marks that topic as "tripped"

3. Cooldown begins: A cooldown counter is set for the tripped topic (default: 10 sweeps)

4. Subsequent sweeps: On each sweep, the cooldown counter decrements for all tripped topics

5. Recovery: When the cooldown reaches zero, the topic is removed from the tripped list

6. Retry: The topic becomes available for publishing attempts again

Benefits

• Prevents blocking: Healthy topics continue to be processed even when some topics fail

• Automatic recovery: Topics automatically recover after the cooldown period

• Resource efficiency: Avoids wasting resources on repeated failures to unhealthy topics

• Observability: Tripped topics can be monitored and alerted on

Configuration

Enabling Circuit Breaking

To enable Sweeper Circuit Breaking, register an IAmAnOutboxCircuitBreaker implementation with your IoC container:

using Paramore.Brighter.CircuitBreaker;

public void ConfigureServices(IServiceCollection services)
{
    // Register the circuit breaker
    services.AddSingleton<IAmAnOutboxCircuitBreaker>(
        new InMemoryOutboxCircuitBreaker(new OutboxCircuitBreakerOptions
        {
            CooldownCount = 10  // Number of sweeps before recovery (default: 10)
        })
    );

    services.AddBrighter(options =>
    {
        // Configure Brighter as normal
    })
    .AddProducers(/* producer configuration */)
    .UseOutboxSweeper();  // Enable the Outbox Sweeper
}

Configuration Options

The OutboxCircuitBreakerOptions class provides the following configuration:

Option	Type	Default	Description
CooldownCount	int	10	Number of sweeper iterations before a tripped topic is eligible for retry

Calculating Cooldown Time

The actual cooldown time depends on your Outbox Sweeper configuration:

Formula: Cooldown Time = CooldownCount × SweepInterval

Example:

• CooldownCount = 10

• Sweeper runs every 60 seconds

• Cooldown Time = 10 × 60s = 10 minutes

services.AddBrighter(options =>
{
    // Sweeper configuration
    options.OutboxSweeper = new OutboxSweeperOptions
    {
        SweepInterval = TimeSpan.FromSeconds(60)  // Sweep every 60 seconds
    };
})
.UseOutboxSweeper();

// Circuit breaker with 10 cooldown sweeps = 10 minutes total cooldown
services.AddSingleton<IAmAnOutboxCircuitBreaker>(
    new InMemoryOutboxCircuitBreaker(new OutboxCircuitBreakerOptions
    {
        CooldownCount = 10  // 10 sweeps × 60s = 10 minutes
    })
);

Usage Patterns

Basic Setup with Outbox Sweeper

public void ConfigureServices(IServiceCollection services)
{
    // Register circuit breaker
    services.AddSingleton<IAmAnOutboxCircuitBreaker>(
        new InMemoryOutboxCircuitBreaker()  // Uses default cooldown of 10 sweeps
    );

    services.AddBrighter(options =>
    {
        options.OutboxSweeper = new OutboxSweeperOptions
        {
            SweepInterval = TimeSpan.FromMinutes(1),  // Sweep every minute
            BatchSize = 100  // Process up to 100 messages per sweep
        };
    })
    .AddProducers(configure =>
    {
        configure.ProducerRegistry = /* your producer registry */;
    })
    .UseMsSqlOutbox(/* outbox configuration */)
    .UseOutboxSweeper();  // Enable sweeper with circuit breaking
}

Custom Cooldown Configuration

Adjust the cooldown based on your needs:

// Short cooldown for quickly recovering topics
services.AddSingleton<IAmAnOutboxCircuitBreaker>(
    new InMemoryOutboxCircuitBreaker(new OutboxCircuitBreakerOptions
    {
        CooldownCount = 3  // Recover after 3 sweeps
    })
);

// Long cooldown for persistent issues
services.AddSingleton<IAmAnOutboxCircuitBreaker>(
    new InMemoryOutboxCircuitBreaker(new OutboxCircuitBreakerOptions
    {
        CooldownCount = 30  // Recover after 30 sweeps
    })
);

Without Circuit Breaking

If you don't register an IAmAnOutboxCircuitBreaker, the sweeper will continue to attempt publishing to all topics even after failures:

// No circuit breaker registered - all topics always attempted
services.AddBrighter(/* configuration */)
    .UseOutboxSweeper();  // Sweeper without circuit breaking

Monitoring and Observability

Checking Tripped Topics

You can query the circuit breaker to see which topics are currently tripped:

public class OutboxMonitorService
{
    private readonly IAmAnOutboxCircuitBreaker _circuitBreaker;
    private readonly ILogger<OutboxMonitorService> _logger;

    public OutboxMonitorService(
        IAmAnOutboxCircuitBreaker circuitBreaker,
        ILogger<OutboxMonitorService> logger)
    {
        _circuitBreaker = circuitBreaker;
        _logger = logger;
    }

    public void CheckCircuitBreakerStatus()
    {
        var trippedTopics = _circuitBreaker.TrippedTopics;

        if (trippedTopics.Any())
        {
            _logger.LogWarning(
                "Circuit breaker has {Count} tripped topics: {Topics}",
                trippedTopics.Count(),
                string.Join(", ", trippedTopics.Select(t => t.Value))
            );
        }
        else
        {
            _logger.LogInformation("All topics are healthy");
        }
    }
}

Logging and Alerts

Set up monitoring to track circuit breaker events:

public class CircuitBreakerHealthCheck : IHealthCheck
{
    private readonly IAmAnOutboxCircuitBreaker _circuitBreaker;

    public CircuitBreakerHealthCheck(IAmAnOutboxCircuitBreaker circuitBreaker)
    {
        _circuitBreaker = circuitBreaker;
    }

    public Task<HealthCheckResult> CheckHealthAsync(
        HealthCheckContext context,
        CancellationToken cancellationToken = default)
    {
        var trippedTopics = _circuitBreaker.TrippedTopics.ToList();

        if (!trippedTopics.Any())
        {
            return Task.FromResult(HealthCheckResult.Healthy("No tripped topics"));
        }

        var data = new Dictionary<string, object>
        {
            { "tripped_topics", trippedTopics.Select(t => t.Value).ToArray() },
            { "count", trippedTopics.Count }
        };

        return Task.FromResult(
            HealthCheckResult.Degraded(
                $"{trippedTopics.Count} topic(s) are currently tripped",
                data: data
            )
        );
    }
}

// Register health check
services.AddHealthChecks()
    .AddCheck<CircuitBreakerHealthCheck>("outbox_circuit_breaker");

Transport-Specific Integration

Brighter V10 includes circuit breaking integration with specific transports:

MongoDB Transport

Circuit breaking is fully integrated with MongoDB Outbox:

services.AddBrighter(/* configuration */)
    .UseMongoDbOutbox(/* MongoDB configuration */)
    .UseOutboxSweeper();

// Circuit breaker works automatically with MongoDB transport
services.AddSingleton<IAmAnOutboxCircuitBreaker>(
    new InMemoryOutboxCircuitBreaker()
);

Other Transports

Circuit breaking works with all Brighter Outbox implementations:

• MS SQL Server (UseMsSqlOutbox)

• PostgreSQL (UsePostgreSqlOutbox)

• MySQL (UseMySqlOutbox)

• SQLite (UseSqliteOutbox)

• DynamoDB (UseDynamoDbOutbox)

• MongoDB (UseMongoDbOutbox)

Bulk Dispatch Support

V10 includes proper circuit breaking support for bulk dispatch operations. When dispatching multiple messages in a batch:

1. Batch grouping: Messages are grouped by topic

2. Per-topic circuit breaking: Each topic's circuit breaker status is checked before dispatching

3. Healthy topics proceed: Only topics that aren't tripped are dispatched

4. Individual retry: Failed batches can be retried individually per topic

// Bulk dispatch respects circuit breaker state
await commandProcessor.ClearOutboxAsync(
    messageIds,  // List of message IDs to dispatch
    continueOnCapturedContext: false,
    cancellationToken: cancellationToken
);

Best Practices

1. Choose Appropriate Cooldown Periods

Balance between quick recovery and avoiding repeated failures:

• Short cooldown (3-5 sweeps): For transient issues, quick recovery desired

• Medium cooldown (10-15 sweeps): General purpose, good balance

• Long cooldown (20-30 sweeps): For persistent issues, reduce retry overhead

2. Align Cooldown with Sweep Interval

Consider the total cooldown time:

// Fast sweeping with short cooldown = quick recovery
options.OutboxSweeper = new OutboxSweeperOptions
{
    SweepInterval = TimeSpan.FromSeconds(30)  // 30s sweep
};

services.AddSingleton<IAmAnOutboxCircuitBreaker>(
    new InMemoryOutboxCircuitBreaker(new OutboxCircuitBreakerOptions
    {
        CooldownCount = 5  // 5 × 30s = 2.5 minutes total cooldown
    })
);

3. Monitor Tripped Topics

Set up monitoring and alerting:

• Health checks: Use ASP.NET Core health checks to expose tripped topics

• Metrics: Export circuit breaker metrics to Prometheus, DataDog, etc.

• Logging: Log when topics trip and recover

• Alerts: Alert when topics remain tripped for extended periods

4. Investigate Root Causes

When topics trip repeatedly:

1. Check broker health: Ensure message broker is operational

2. Verify permissions: Ensure the application has permissions to publish

3. Check queue/topic existence: Verify the destination exists

4. Review capacity: Check if the queue/topic has reached capacity limits

5. Inspect network: Look for network connectivity issues

5. Use with Outbox Sweeper

Circuit breaking is designed to work with the Outbox Sweeper:

// Always enable UseOutboxSweeper when using circuit breaking
services.AddBrighter(/* configuration */)
    .UseOutboxSweeper()  // Required for circuit breaking to function
    .UseMsSqlOutbox(/* outbox config */);

6. Consider Immediate vs. Sweeper Clearing

Circuit breaking only applies to sweeper-based clearing:

// Immediate clearing - NOT subject to circuit breaking
await postBox.ClearOutboxAsync(messageIds);

// Sweeper clearing - subject to circuit breaking
// Happens automatically via UseOutboxSweeper

7. Test Failure Scenarios

Regularly test circuit breaker behavior:

• Simulate broker outages

• Test individual topic failures

• Verify healthy topics continue processing

• Confirm automatic recovery after cooldown

Troubleshooting

Topics Not Recovering

Problem: Topics remain tripped indefinitely

Solutions:

1. Verify Outbox Sweeper is running

2. Check cooldown count is not excessively high

3. Ensure sweeper interval is appropriate

4. Confirm circuit breaker is properly registered

All Topics Tripping

Problem: All topics become tripped at once

Possible Causes:

• Broker is completely down

• Network connectivity issues

• Authentication/authorization failures

• Shared resource exhaustion

Solutions:

1. Check broker health and connectivity

2. Verify credentials and permissions

3. Review broker logs for errors

4. Consider infrastructure capacity

Messages Stuck in Outbox

Problem: Messages accumulate in Outbox without being dispatched

Check:

1. Is the Outbox Sweeper enabled?

2. Are topics currently tripped? Check TrippedTopics

3. Is the circuit breaker cooldown too long?

4. Are there persistent transport issues?

Solutions:

• Enable Outbox Sweeper if not already enabled

• Investigate why topics are tripping

• Reduce cooldown count if appropriate

• Fix underlying transport issues

Circuit Breaker Not Working

Problem: Failed topics continue to be retried

Verify:

1. Circuit breaker is registered: services.AddSingleton<IAmAnOutboxCircuitBreaker>

2. Using the sweeper: UseOutboxSweeper()

3. Exceptions are being thrown during publish (not silently failing)

4. Circuit breaker implementation is correct

Advanced Scenarios

Custom Circuit Breaker Implementation

Implement IAmAnOutboxCircuitBreaker for custom behavior:

public class CustomOutboxCircuitBreaker : IAmAnOutboxCircuitBreaker
{
    private readonly Dictionary<RoutingKey, CircuitBreakerState> _topics = new();

    public void TripTopic(RoutingKey topic)
    {
        _topics[topic] = new CircuitBreakerState
        {
            TrippedAt = DateTime.UtcNow,
            FailureCount = _topics.ContainsKey(topic)
                ? _topics[topic].FailureCount + 1
                : 1
        };

        // Custom logic: Log, emit metrics, send alerts, etc.
    }

    public void CoolDown()
    {
        var now = DateTime.UtcNow;
        var recovered = new List<RoutingKey>();

        foreach (var kvp in _topics)
        {
            var cooldownPeriod = TimeSpan.FromMinutes(10);
            if (now - kvp.Value.TrippedAt > cooldownPeriod)
            {
                recovered.Add(kvp.Key);
            }
        }

        foreach (var topic in recovered)
        {
            _topics.Remove(topic);
            // Custom logic: Log recovery, emit metrics, etc.
        }
    }

    public IEnumerable<RoutingKey> TrippedTopics => _topics.Keys;
}

Distributed Circuit Breaker

For multi-instance deployments, consider a distributed circuit breaker using Redis, SQL, or other shared storage:

public class DistributedOutboxCircuitBreaker : IAmAnOutboxCircuitBreaker
{
    private readonly IDistributedCache _cache;

    public void TripTopic(RoutingKey topic)
    {
        var key = $"circuit-breaker:{topic.Value}";
        _cache.SetString(key, DateTime.UtcNow.ToString(), new DistributedCacheEntryOptions
        {
            AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(10)
        });
    }

    // Implement other methods using distributed cache
}

Summary

Sweeper Circuit Breaking provides automatic resilience for Outbox clearing operations by:

• Preventing cascade failures when specific topics fail

• Automatically recovering after a configurable cooldown period

• Allowing healthy topics to continue processing

• Protecting resources from repeated failures

Enable circuit breaking by registering IAmAnOutboxCircuitBreaker with your IoC container and configuring appropriate cooldown periods for your application's needs.