9
\$\begingroup\$

I wrote a simple utility for Retry and CircuitBreaker patterns. I implemented it by using ReactiveX because I thought it might actually be the perfect framework for it. I know that there is a project called Polly but it didn't convince me and I wanted to try Rx out.

My little tool can basically do two things:

  • It can retry an action as many times as I want and wait in between the specified amount of time.
  • It can break retries of one or multiple retry handlers with a circuit-breaker.

Retry

The Retry class implements the IObservable<Exception> interface. It notifies its observers (if any) every time the action that might fail actually failed by sending it the current exception via OnNext. Retry never calls OnCompleted because it actually never completes and we don't want to unsubscribe any observables in case we reuse the same Retry another time.

Internally it maintains a HashSet of CircutBreakers that provide a Token property that is necessary to cancel the Retry.

I'm not sure whether I should actually collect the exceptions an throw the AggregateException at the end or just leave the method with a failed Task and let other observer(s) take care of (maybe log) the exception(s).

public class Retry : IObservable<Exception>
{
    private readonly IEnumerable<TimeSpan> _delays;

    private readonly Subject<Exception> _subject;

    private readonly HashSet<CircutBraker> _circutBreakers;

    private Retry(IEnumerable<TimeSpan> delays)
    {
        // Deliberately not materializing the delays. We don't know whether this is not infinite.
        _delays = delays;
        _subject = new Subject<Exception>();
        _circutBreakers = new HashSet<CircutBraker>();
    }

    public IDisposable Subscribe(IObserver<Exception> exceptionObserver)
    {
        if (exceptionObserver is CircutBraker circutBreaker)
        {
            _circutBreakers.Add(circutBreaker);
            var circutBreakerUnsubscriber = _subject.Subscribe(exceptionObserver);
            return Disposable.Create(() =>
            {
                _circutBreakers.Remove(circutBreaker);
                circutBreakerUnsubscriber.Dispose();
            });
        }
        else
        {
            return _subject.Subscribe(exceptionObserver);
        }
    }

    public static Retry Create(IEnumerable<TimeSpan> delays)
    {
        return new Retry(delays);
    }

    public async Task<T> TryExecuteAsync<T>(Func<CancellationToken, Task<T>> action, CancellationToken cancellationToken)
    {
        var exceptions = new List<Exception>();

        foreach (var delay in _delays)
        {
            ThrowIfCancellationRequested();

            try
            {
                return await action(cancellationToken);
            }
            catch (Exception ex)
            {
                exceptions.Add(ex);
                _subject.OnNext(ex);

                // Don't delay if cancelled but return immediately.
                ThrowIfCancellationRequested();

                await Task.Delay(delay);
            }
        }

        throw new AggregateException(exceptions);

        void ThrowIfCancellationRequested()
        {
            foreach (var cb in _circutBreakers)
            {
                cb.Token.ThrowIfCancellationRequested();
            }
        }
    }
}

CircutBreaker v1

The CircutBreaker class is implemented on top of the IObserver<Exception> interface and the main logic is created inside the constructor. Currently it supports just two states: Closed and Opened - because I don't have any scenario yet where I could use the third state and check if the error is gone (I might implement it later) - thus the HalfOpen case is empty.

When the number of failures exceeds the allowed count within the specified interval the CancellationTokenSource is cancelled. Retry watches for this token and reacts accordingly by checking the Token property - this allows me to cancel all retries immediately.

public class CircutBreaker : IObserver<Exception>, IDisposable
{
    private readonly IObserver<Exception> _observer;

    private DateTime _lastFailureOn;

    private int _failureCount;

    private CircutBrakerState _state = CircutBrakerState.Closed;

    private CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();

    private CircutBreaker(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
    {       
        _observer = Observer.Create<Exception>(exception =>
        {
            if (filter?.Invoke(exception) == false)
            {
                return;
            }

            var now = DateTime.UtcNow;

            switch (_state)
            {
                case CircutBrakerState.Closed:
                    {
                        var isFailureWithinInterval = now - _lastFailureOn < maxFailureInterval;
                        var canCountFailure = isFailureWithinInterval || _failureCount == 0;

                        _failureCount = canCountFailure ? _failureCount + 1 : 0;

                        if (_failureCount == maxFailureCount)
                        {
                            _state = CircutBrakerState.Open;
                            _cancellationTokenSource.Cancel();
                            Debug.WriteLine("Circut broken!");
                        }

                        _lastFailureOn = now;
                    }
                    break;

                case CircutBrakerState.HalfOpen:
                    // I currently don't have any use case for this.
                    break;

                case CircutBrakerState.Open:
                    // There is nothing to do when Open.
                    break;
            }
        });
    }

    public CircutBrakerState State => _state;

    public CancellationToken Token => _cancellationTokenSource.Token;

    public void OnNext(Exception exception) => _observer.OnNext(exception);

    public void OnCompleted() => _observer.OnCompleted();

    public void OnError(Exception exception) => _observer.OnError(exception);

    public void Reset()
    {
        _lastFailureOn = DateTime.MinValue;
        _failureCount = 0;
        _state = CircutBrakerState.Closed;
        _cancellationTokenSource.Dispose();
        _cancellationTokenSource = new CancellationTokenSource();
    }

    public void Dispose() => _cancellationTokenSource.Dispose();

    public static CircutBreaker Create(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
    {
        return new CircutBreaker(maxFailureCount, maxFailureInterval, filter);
    }
}

CircutBreaker v2

There is also an alternative CircutBreaker constructor which uses a new extension to open the circut:

private CircutBreaker(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
{
    var openFuse = Observer.Create<Exception>(exception =>
    {
        switch (_state)
        {
            case CircutBrakerState.Closed:
                _state = CircutBrakerState.Open;
                _cancellationTokenSource.Cancel();
                Debug.WriteLine("Circut broken!");
                break;

            case CircutBrakerState.HalfOpen:
                // I currently don't have any use case for this.
                break;

            case CircutBrakerState.Open:
                // There is nothing to do when Open.
                break;
        }
    });

    _observer = new Subject<Exception>();
    ((IObservable<Exception>)_observer)
        .Where(ex => filter == null || filter(ex))
        .Fuse(maxFailureCount, maxFailureInterval)
        .Subscribe(openFuse);
}

I named the extension Fuse because it lets only the last item through that occurred within the specified interval:

public static class ObservableExtensions
{
    public static IObservable<T> Fuse<T>(this IObservable<T> observable, int count, TimeSpan interval)
    {
        var currentCount = 0;
        var lastCountOn = DateTime.MinValue;

        var fuse = new Subject<T>();

        observable.Subscribe(Observer.Create<T>(item =>
        {
            if (currentCount == 0 || DateTime.UtcNow - lastCountOn < interval)
            {
                currentCount++;
                if (currentCount == count)
                {
                    fuse.OnNext(item);
                    currentCount = 0;
                }
            }
            else
            {
                currentCount = 0;
            }

            lastCountOn = DateTime.UtcNow;
        }));

        return fuse;
    }
}

I wrote this one in order to remove the fuse logic out of the CircuitBreaker. Do you think it is a good choice? This is why I added the tag .


The states are defined in an enum:

public enum CircutBreakerState
{
    Closed,
    HalfOpen,
    Open,
}

I also crated an observer-logger that I can use to log exceptions or display the message to the console:

public class ExceptionLogger
{
    public static IObserver<Exception> Create(Action<Exception> logException)
    {
        return Observer.Create(logException);
    }
}

Example

I tested it by creating two Retry objects and one CircuitBreaker. I then attached it to both retries and executed a method that fails. Additionally, I attached a logger to each retry.

var retry1 = Retry.Create(new[]
{
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3)
});

var retry2 = Retry.Create(new[]
{
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3),
    TimeSpan.FromSeconds(3)
});

var circutBreaker = CircutBreaker.Create(2, TimeSpan.FromSeconds(6));

retry1.Subscribe(ExceptionLogger.Create(ex => Console.WriteLine("Log: " + ex.Message)));
retry2.Subscribe(ExceptionLogger.Create(ex => Console.WriteLine("Log: " + ex.Message)));
retry1.Subscribe(circutBreaker);
retry2.Subscribe(circutBreaker);


var task1 = retry1.TryExecuteAsync(async (cancellationToken) =>
{
    Console.WriteLine("1st method called.");
    await Task.Delay(TimeSpan.FromSeconds(0.5));
    throw new Exception("1st method failed.");
    return "Test";
}, CancellationToken.None);

var task2 = retry2.TryExecuteAsync(async (cancellationToken) =>
{
    Console.WriteLine("2nd method called.");
    await Task.Delay(TimeSpan.FromSeconds(0.5));
    throw new Exception("2nd method failed.");
    return "Test";
}, CancellationToken.None);

Task.WaitAll(new[] { task1, task2 });

The output is:

1st method called.
2nd method called.
Log: 2nd method failed.
Log: 1st method failed.
Circut broken!

It works very nice but I think it can still be made better and more reactive.

\$\endgroup\$
  • \$\begingroup\$ looks interesting. do you have tests for this as well? - if you could post that as well, that would be handy - or even better to link the relevant classes to github? \$\endgroup\$ – BKSpurgeon Oct 23 '17 at 11:46
  • \$\begingroup\$ @BKSpurgeon currently this exists only as an experiment in my LINQPad. I was going to check this in after getting some reviews, cleaning it up and tuning ;-) This is my first time with Rx and I'm not sure whether I use it as it supposed to be used. It works but it'll be nice to get a confirmation or feedback how to do it correctly (in case the code violates some Rx principles). \$\endgroup\$ – t3chb0t Oct 23 '17 at 11:50
2
\$\begingroup\$

Ease of Use:

This seems a little clumsy to use. I have to build an IEnumerable<TimeSpan> every time I use it. That's great if I want to retry with a small set of times or I want to have the delays follow a logarithmic curve to reduce the possibility of DDoS'ing whatever I'm calling. However, it makes it a pain to use if I want it to try N times with the same timeout where N > 5. At this point, I'd probably construct a method to handle this, which would be one more thing for me to have to test. If I wrote this, I'd make a second constructor that took a TimeSpan and an optional number of times to retry (no value provided would mean retry forever).

I wouldn't go to this extent for purely personal code, but if it was a library, I'd probably make the previous constructor take a Func<T, int, TimeSpan> argument that calculated the time to wait based on the value passed in and the number of times retried instead of using the count of delays being the number to retry.

Best Practices:

As a general rule, you don't want to be calling .Dispose() on an IDisposable. Use a using block.

Watch out for Bugs:

_cancellationTokenSource.Dispose();
_cancellationTokenSource = new CancellationTokenSource();

First, you aren't cancelling this (what if someone Resets it when it is Closed? They will if you let them). If someone Resets the CircuitBreaker when it is being used, what's going to happen when something checks to see if cancellation has been requested? The cancellation token source you are using has been disposed, right? Well, due to reference-type semantics you actually just gave overwrote the old instance, which is presumably now free for garbage collection, with the newed-up new CancellationTokenSource. Now everything that references this actually just starts using the new cancellation token source. Believe me when I say that this can be the root of an interesting, confusing bug if you aren't thinking; I had it eat several hours of time once.

Currently if you Dispose a cancellation token source without Cancelling it it reports that cancellation was not requested. This is unlikely to change given it has been the given behavior for so long, but may or may not be part of the specification. If this was me, I'd check to see if the state is Closed and if it is, not do anything if a Reset is requested. You may need to have more custom handling here if/when you implement HalfClosed as well.

\$\endgroup\$
  • \$\begingroup\$ it makes it a pain to use if I want it to try N times with the same timeout where N > 5. At this point, I'd probably construct a method to handle this - why? Using enumerables is very easy and convenient e.g. you could use Enumerable.Repeat to get 5 delays. if it was a library, I'd probably make the previous constructor take [...] - I cannot agree on that one either. It's not the responsibility of the Retry class to generate any delay sequences. An enumerable is all it needs. Any other logic should be encapsulated in another specialized class like DelayGenerator etc. \$\endgroup\$ – t3chb0t Mar 24 '18 at 14:50
  • \$\begingroup\$ I just don't like relying on the length of the enumerable to set the number of retries. It's very non-obvious, which makes it hard to debug. I like the ability to generate delays of different lengths, but I'd prefer it was obvious at a glance how many times it would retry. For example, someone could move the generation of the enumerable to a method with a hard-coded length, and the maintainer wouldn't necessarily look into that method and be confused. \$\endgroup\$ – Hosch250 Mar 24 '18 at 14:58
  • \$\begingroup\$ someone could move the generation of the enumerable to a method with a hard-coded length - then I say it's his fault ;-) Trying to prevent other developer's mistakes by making your code unnecessarily more difficult to test isn't a good option. You can add helper classes (like the DelayGenerator) to the library but the core should not take damage because there is someone who cannot write clean code ;-] Well, anyway, I addressed a couple of the issues you mentioned and I will post a self-answer later... \$\endgroup\$ – t3chb0t Mar 24 '18 at 15:04
0
\$\begingroup\$

Reactive programming is fun but sometimes it can be a real mind-fuck. Improving my Retry and CircuitBreaker wasn't easy but this is what I've got now. I'm still trying to understand what I'm doing, especially the cancellation so I'm not claiming it's perfect. However, I hope it's at least a little bit better then the last version.


I simplified the Retry class by removing the internal Subject and the dependency on the CircuitBreaker. It doesn't know anything about it anymore. The cancellation works with a TryExecuteAsync local CancellationTokenSource.

public class Retry : IObservable<Attempt>
{
    private readonly IEnumerable<TimeSpan> _delays;

    private readonly ISet<IObserver<Attempt>> _attemptRxs;

    private Retry(IEnumerable<TimeSpan> delays)
    {
        // Not materializing the delays because it might be infinite.
        _delays = delays;
        _attemptRxs = new HashSet<IObserver<Attempt>>();
    }

    public static Retry Create(IEnumerable<TimeSpan> delays)
    {
        if (delays == null) throw new ArgumentNullException(paramName: nameof(delays));
        return new Retry(delays);
    }

    public IDisposable Subscribe(IObserver<Attempt> attemptRx)
    {
        if (attemptRx == null) throw new ArgumentNullException(paramName: nameof(attemptRx));
        if (_attemptRxs.Add(attemptRx))
        {
            return Disposable.Create(() =>
            {
                _attemptRxs.Remove(attemptRx);
            });
        }
        else
        {
            throw new InvalidOperationException("Observer already subscribed.");
        }
    }

    public async Task<T> TryExecuteAsync<T>(Func<CancellationToken, Task<T>> action, CancellationToken cancellationToken)
    {
        var exceptions = new List<Exception>();

        using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
        {
            var attempts = 0;
            foreach (var delay in _delays)
            {
                attempts++;
                Debug.WriteLine($"Attempt {attempts}.");

                try
                {
                    return await action(cts.Token);
                }
                catch (Exception ex)
                {
                    exceptions.Add(ex);
                    foreach (var exceptionRx in _attemptRxs)
                    {
                        exceptionRx.OnNext(new Attempt(attempts, ex, cts.Cancel));
                    }

                    Debug.WriteLineIf(cts.IsCancellationRequested, "Cancelled.");

                    // Don't delay but cancel immediately.
                    cts.Token.ThrowIfCancellationRequested();

                    await Task.Delay(delay);
                }
            }
        }

        throw new AggregateException(exceptions);
    }
}

I replaced the CircuitBreaker dependency by the new Attempt class that any observer can receive and call the Cancel method on it in order to cancel the retry process.

public class Attempt
{
    private Action _cancel;

    public Attempt(int count, Exception excepiton, Action cancel)
    {
        Count = count;
        Exception = excepiton;
        _cancel = cancel;
    }

    public int Count { get; }

    public Exception Exception { get; set; }

    public void Cancel() => _cancel();
}

The CircuitBreaker is now calling the Cancel method of the Attempt object if the Fuse is blown. In the open state the fuse is disabled.

public class CircuitBreaker : IObserver<Attempt>, IDisposable
{
    private ISubject<Attempt> _subject;
    private IDisposable _unsubscriber;

    public CircuitBreaker(int maxFailureCount, TimeSpan maxFailureInterval)
    {
        if (maxFailureCount <= 0) throw new ArgumentOutOfRangeException(paramName: nameof(maxFailureCount), message: $"{maxFailureCount} must be greater than 0.");
        if (maxFailureInterval.Ticks == 0) throw new ArgumentOutOfRangeException(paramName: nameof(maxFailureInterval), message: $"{maxFailureInterval} must be greater than 0.");

        MaxFailureCount = maxFailureCount;
        MaxFailureInterval = maxFailureInterval;

        _subject = new Subject<Attempt>();
        _unsubscriber =
            _subject
                .Fuse(maxFailureCount, maxFailureInterval, enabled: () => State == CircutBrakerState.Closed)
                .Subscribe(failure =>
                {                   
                    State = CircutBrakerState.Open;
                    failure.Cancel();
                    Debug.WriteLine($"{nameof(CircuitBreaker)} {State}.");
                });
    }

    public int MaxFailureCount { get; }

    public TimeSpan MaxFailureInterval { get; }

    public CircutBrakerState State { get; private set; } = CircutBrakerState.Closed;

    public void Reset()
    {
        State = CircutBrakerState.Closed;
    }

    public void OnNext(Attempt attempt)
    {
        _subject.OnNext(attempt);
    }

    public void OnCompleted()
    {
        _subject.OnCompleted();
    }

    public void OnError(Exception error)
    {
        _subject.OnError(error);
    }

    public void Dispose()
    {
        _unsubscriber.Dispose();
    }
}

The implementation of the Fuse extension changed slightly too because it now has the new enabled parameter. Without it the CircuitBreaker couldn't cancel other tasks because the fuse would not pass any messages through.

public static IObservable<T> Fuse<T>(this IObservable<T> observable, int count, TimeSpan interval, Func<bool> enabled)
{
    if (observable == null) throw new ArgumentNullException(paramName: nameof(observable));
    if (count <= 0) throw new ArgumentOutOfRangeException(paramName: nameof(count), message: $"{count} must be greater than 0.");
    if (interval.Ticks == 0) throw new ArgumentOutOfRangeException(paramName: nameof(interval), message: $"{interval} must be greater than 0.");
    if (enabled == null) throw new ArgumentNullException(paramName: nameof(enabled));

    Debug.WriteLine($"Fuse threshold is {count} in {interval:mm\\:ss\\:fff}.");

    var occuranceCount = 0;
    var lastOccuranceStopwatch = Stopwatch.StartNew();

    return Observable.Create<T>(x => observable.Subscribe(Observer.Create<T>(item =>
    {
        if (enabled())
        {
            if (++occuranceCount == count)
            {
                if (lastOccuranceStopwatch.Elapsed <= interval)
                {
                    Debug.WriteLine($"Fuse blown at {count} in {lastOccuranceStopwatch.Elapsed:mm\\:ss\\:fff} of {interval:mm\\:ss\\:fff}.");
                    x.OnNext(item);
                }

                Reset();
            }

            if (lastOccuranceStopwatch.Elapsed > interval)
            {
                Reset();
            }
        }
        else
        {
            x.OnNext(item);
        }
    })));

    void Reset()
    {
        Debug.WriteLine($"Fuse reset at {occuranceCount} in {lastOccuranceStopwatch.Elapsed:mm\\:ss\\:fff}.");
        occuranceCount = 0;
        lastOccuranceStopwatch.Restart();
    }
}

I also added a new class that I call Delay that contains a couple of convenience methods for generating delays. This is one of them:

public static class Delay
{
    public static IEnumerable<TimeSpan> Repeat(TimeSpan delay, int? count = null)
    {
        var i = 0;
        while (!count.HasValue || (i < count.Value))
        {
            i++;
            yield return delay;
        }
    }
}

Example

The usage did not change much because most of the changes were internal ones but for the sake of completnes here's the new use case:

void Main()
{    
    var retry1 = Retry.Create(Delay.Repeat(TimeSpan.FromSeconds(2), 10));
    var retry2 = Retry.Create(Delay.Repeat(TimeSpan.FromSeconds(3), 10));

    var cb = new CircuitBreaker(3, TimeSpan.FromSeconds(5));

    retry1.Subscribe(attempt => attempt.Exception.Message.Dump());
    retry2.Subscribe(attempt => attempt.Exception.Message.Dump());
    retry1.Subscribe(cb);
    retry2.Subscribe(cb);

    var task1 = retry1.TryExecuteAsync(async (cancellationToken) =>
    {
        Console.WriteLine($"Task-1 called [{Thread.CurrentThread.ManagedThreadId}].");
        await Task.Delay(TimeSpan.Zero);
        throw new Exception("Task-1 failed.");
        return "Test";
    }, CancellationToken.None);

    var task2 = retry2.TryExecuteAsync(async (cancellationToken) =>
    {
        Console.WriteLine($"Task-2 called [{Thread.CurrentThread.ManagedThreadId}].");
        await Task.Delay(TimeSpan.FromSeconds(2));
        throw new Exception("Task-2 failed.");
        return "Test";
    }, CancellationToken.None);

    Task.WaitAll(new[] { task1, task2 });

}

Output:

Fuse threshold is 3 in 00:05:000.
Attempt 1.
Task-1 called [20].
Task-1 failed.
Attempt 1.
Task-2 called [20].
Attempt 2.
Task-1 called [6].
Task-1 failed.
Task-2 failed.
Fuse blown at 3 in 00:02:006 of 00:05:000.
CircuitBreaker Open.
Fuse reset at 3 in 00:02:007.
Cancelled.
Attempt 3.
Task-1 called [6].
Task-1 failed.
CircuitBreaker Open.
Cancelled.
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.