feat(circuit-breaker): add job ID + old/new state change to circuit-breaker transition (#469)

stijnmoreels · web-flow · commit f38dc6488621 · 2025-02-10T12:41:01.000+01:00
* feat: add job ID + old/new state change to circuit-breaker transition

* pr-sug: rename state change to '...EventArgs'

* pr-sug: rename event-args to `MessagePumpCircuitStateChangedEventArgs`
diff --git a/docs/preview/02-Features/06-general-messaging.md b/docs/preview/02-Features/06-general-messaging.md
@@ -96,9 +96,16 @@ The instances should implement the `ICircuitBreakerEventHandler`, which allows y
 ```csharp
 public class MyFirstCircuitBreakerEventHandler : ICircuitBreakerEventHandler
 {
-    public Task OnTransitionAsync(MessagePumpCircuitState newState)
+    public Task OnTransitionAsync(MessagePumpCircuitStateChang change)
     {
-        // ...
+        // The job ID of the message pump that was transitioned.
+        string jobId = change.JobId;
+
+        // The circuit breaker state transitions.
+        MessagePumpCircuitState oldState = change.OldState;
+        MessagePumpCircuitState newState = change.NewState;
+
+        // Process the state change event...
     }
 }
 ```
diff --git a/src/Arcus.Messaging.Pumps.Abstractions/MessagePump.cs b/src/Arcus.Messaging.Pumps.Abstractions/MessagePump.cs
@@ -145,9 +145,12 @@ protected async Task WaitMessageRecoveryPeriodAsync(CancellationToken cancellati
 
             if (!CircuitState.IsHalfOpen)
             {
-                CircuitState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);
+                MessagePumpCircuitState
+                    oldState = CircuitState,
+                    newState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);
 
-                NotifyCircuitBreakerStateChangedSubscribers();
+                CircuitState = newState;
+                NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);
             }
         }
 
@@ -162,9 +165,12 @@ protected async Task WaitMessageIntervalDuringRecoveryAsync(CancellationToken ca
 
             if (!CircuitState.IsHalfOpen)
             {
-                CircuitState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);
+                MessagePumpCircuitState
+                    oldState = CircuitState,
+                    newState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);
 
-                NotifyCircuitBreakerStateChangedSubscribers();
+                CircuitState = newState;
+                NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);
             }
         }
 
@@ -176,9 +182,12 @@ internal void NotifyPauseReceiveMessages(MessagePumpCircuitBreakerOptions option
         {
             Logger.LogDebug("Circuit breaker caused message pump '{JobId}' to transition from a '{CurrentState}' an 'Open' state", JobId, CircuitState);
 
-            CircuitState = CircuitState.TransitionTo(CircuitBreakerState.Open, options);
+            MessagePumpCircuitState
+                oldState = CircuitState,
+                newState = CircuitState.TransitionTo(CircuitBreakerState.Open, options);
 
-            NotifyCircuitBreakerStateChangedSubscribers();
+            CircuitState = newState;
+            NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);
         }
 
         /// <summary>
@@ -188,18 +197,21 @@ protected void NotifyResumeRetrievingMessages()
         {
             Logger.LogDebug("Circuit breaker caused message pump '{JobId}' to transition back from '{CurrentState}' to a 'Closed' state, retrieving messages is resumed", JobId, CircuitState);
 
-            CircuitState = MessagePumpCircuitState.Closed;
+            MessagePumpCircuitState
+                oldState = CircuitState,
+                newState = MessagePumpCircuitState.Closed;
 
-            NotifyCircuitBreakerStateChangedSubscribers();
+            CircuitState = newState;
+            NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);
         }
 
-        private void NotifyCircuitBreakerStateChangedSubscribers()
+        private void NotifyCircuitBreakerStateChangedSubscribers(MessagePumpCircuitState oldState, MessagePumpCircuitState newState)
         {
             ICircuitBreakerEventHandler[] eventHandlers = GetEventHandlersForPump();
 
             foreach (var handler in eventHandlers)
             {
-                Task.Run(() => handler.OnTransition(CircuitState));
+                Task.Run(() => handler.OnTransition(new MessagePumpCircuitStateChangedEventArgs(JobId, oldState, newState)));
             }
         }
 
diff --git a/src/Arcus.Messaging.Pumps.Abstractions/Resiliency/IMessagePumpCircuitBreaker.cs b/src/Arcus.Messaging.Pumps.Abstractions/Resiliency/IMessagePumpCircuitBreaker.cs
@@ -33,8 +33,47 @@ public interface ICircuitBreakerEventHandler
         /// <summary>
         /// Notifies the application on a change in the message pump's circuit breaker state.
         /// </summary>
-        /// <param name="newState">The new circuit breaker state in which the message pump is currently running on.</param>
-        void OnTransition(MessagePumpCircuitState newState);
+        /// <param name="args">The change in the circuit breaker state for a message pump.</param>
+        void OnTransition(MessagePumpCircuitStateChangedEventArgs args);
+    }
+
+    /// <summary>
+    /// Represents a change event of the <see cref="MessagePumpCircuitState"/> in a <see cref="MessagePump"/>.
+    /// </summary>
+    public class MessagePumpCircuitStateChangedEventArgs
+    {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="MessagePumpCircuitStateChangedEventArgs" /> class.
+        /// </summary>
+        internal MessagePumpCircuitStateChangedEventArgs(
+            string jobId,
+            MessagePumpCircuitState oldState,
+            MessagePumpCircuitState newState)
+        {
+            if (string.IsNullOrWhiteSpace(jobId))
+            {
+                throw new ArgumentException("Requires a non-blank job ID for the circuit breaker event state change registration", nameof(jobId));
+            }
+
+            JobId = jobId;
+            OldState = oldState;
+            NewState = newState;
+        }
+
+        /// <summary>
+        /// Gets the unique ID to distinguish the linked message pump that had it circuit breaker state changed.
+        /// </summary>
+        public string JobId { get; }
+
+        /// <summary>
+        /// Gets the original circuit breaker state the linked message pump was in.
+        /// </summary>
+        public MessagePumpCircuitState OldState { get; }
+
+        /// <summary>
+        /// Gets the current circuit breaker state the linked message pump is in.
+        /// </summary>
+        public MessagePumpCircuitState NewState { get; }
     }
 
     /// <summary>
diff --git a/src/Arcus.Messaging.Tests.Integration/MessagePump/Fixture/MockCircuitBreakerEventHandler.cs b/src/Arcus.Messaging.Tests.Integration/MessagePump/Fixture/MockCircuitBreakerEventHandler.cs
@@ -1,8 +1,10 @@
-﻿using System.Collections.Generic;
+﻿using System;
 using System.Collections.ObjectModel;
-using System.Linq;
+using System.Threading.Tasks;
 using Arcus.Messaging.Pumps.Abstractions.Resiliency;
+using Arcus.Testing;
 using Xunit;
+using Xunit.Sdk;
 
 namespace Arcus.Messaging.Tests.Integration.MessagePump.Fixture
 {
@@ -11,28 +13,32 @@ namespace Arcus.Messaging.Tests.Integration.MessagePump.Fixture
     /// </summary>
     internal class MockCircuitBreakerEventHandler : ICircuitBreakerEventHandler
     {
-        private readonly Collection<MessagePumpCircuitState> _states = new();
+        private readonly Collection<MessagePumpCircuitStateChangedEventArgs> _states = new();
 
         /// <summary>
         /// Notifies the application on a change in the message pump's circuit breaker state.
         /// </summary>
-        /// <param name="newState">The new circuit breaker state in which the message pump is currently running on.</param>
-        public void OnTransition(MessagePumpCircuitState newState)
+        /// <param name="args">The change in the circuit breaker state for a message pump.</param>
+        public void OnTransition(MessagePumpCircuitStateChangedEventArgs args)
         {
-            _states.Add(newState);
+            _states.Add(args);
         }
 
-        public void ShouldTransitionCorrectly()
+        /// <summary>
+        /// Verifies that all fired circuit breaker state transitions are happening correctly.
+        /// </summary>
+        public async Task ShouldTransitionedCorrectlyAsync()
         {
-            Assert.NotEmpty(_states);
-
-            MessagePumpCircuitState firstTransition = _states[0];
-            Assert.True(firstTransition.IsOpen, $"when the message pump starts up, the first transition should always be from a closed to open state, but got: {firstTransition}");
+            await Poll.Target<XunitException>(() =>
+            {
+                Assert.NotEmpty(_states);
+                Assert.Equal(3, _states.Count);
 
-            IEnumerable<(MessagePumpCircuitState oldState, MessagePumpCircuitState newState)> transitions =
-                _states.SkipLast(1).Zip(_states.Skip(1));
+            }).Every(TimeSpan.FromSeconds(1))
+              .Timeout(TimeSpan.FromSeconds(10))
+              .FailWith("could not in time find all the fired circuit breaker change events, possibly the message pump did not fired them");
 
-            Assert.All(transitions, t => VerifyCorrectTransition(t.oldState, t.newState));
+            Assert.All(_states, t => VerifyCorrectTransition(t.OldState, t.NewState));
         }
 
         private static void VerifyCorrectTransition(
@@ -41,7 +47,7 @@ private static void VerifyCorrectTransition(
         {
             if (oldState.IsClosed)
             {
-                Assert.True(newState.IsHalfOpen, $"when the message pump comes from a closed state, the next state should always be half-open, but got: {newState}");
+                Assert.True(newState.IsOpen, $"when the message pump comes from a closed state, the next state should always be open, but got: {newState}");
             }
             else if (oldState.IsOpen)
             {
diff --git a/src/Arcus.Messaging.Tests.Integration/MessagePump/ServiceBusMessagePump.ResiliencyTests.cs b/src/Arcus.Messaging.Tests.Integration/MessagePump/ServiceBusMessagePump.ResiliencyTests.cs
@@ -62,8 +62,8 @@ public async Task ServiceBusMessageQueuePump_WithUnavailableDependencySystem_Cir
             await producer.ProduceAsync(messageAfterBreak);
             await messageSink.ShouldReceiveOrdersAfterBreakAsync(messageAfterBreak.MessageId);
 
-            mockEventHandler1.ShouldTransitionCorrectly();
-            mockEventHandler2.ShouldTransitionCorrectly();
+            await mockEventHandler1.ShouldTransitionedCorrectlyAsync();
+            await mockEventHandler2.ShouldTransitionedCorrectlyAsync();
         }
 
         [Fact]
@@ -99,8 +99,8 @@ public async Task ServiceBusMessageTopicPump_WithUnavailableDependencySystem_Cir
             await producer.ProduceAsync(messageAfterBreak);
             await messageSink.ShouldReceiveOrdersAfterBreakAsync(messageAfterBreak.MessageId);
 
-            mockEventHandler1.ShouldTransitionCorrectly();
-            mockEventHandler2.ShouldTransitionCorrectly();
+            await mockEventHandler1.ShouldTransitionedCorrectlyAsync();
+            await mockEventHandler2.ShouldTransitionedCorrectlyAsync();
         }
 
         private async Task<TemporaryTopicSubscription> CreateTopicSubscriptionForMessageAsync(params ServiceBusMessage[] messages)

Original file line number	Diff line number	Diff line change
@@ -96,9 +96,16 @@ The instances should implement the `ICircuitBreakerEventHandler`, which allows y
`96`	`96`	```csharp
`97`	`97`	`public class MyFirstCircuitBreakerEventHandler : ICircuitBreakerEventHandler`
`98`	`98`	`{`
`99`		`- public Task OnTransitionAsync(MessagePumpCircuitState newState)`
	`99`	`+ public Task OnTransitionAsync(MessagePumpCircuitStateChang change)`
`100`	`100`	`{`
`101`		`- // ...`
	`101`	`+ // The job ID of the message pump that was transitioned.`
	`102`	`+ string jobId = change.JobId;`
	`103`	`+`
	`104`	`+ // The circuit breaker state transitions.`
	`105`	`+ MessagePumpCircuitState oldState = change.OldState;`
	`106`	`+ MessagePumpCircuitState newState = change.NewState;`
	`107`	`+`
	`108`	`+ // Process the state change event...`
`102`	`109`	`}`
`103`	`110`	`}`
`104`	`111`	```
Original file line number	Diff line number	Diff line change
`@@ -145,9 +145,12 @@ protected async Task WaitMessageRecoveryPeriodAsync(CancellationToken cancellati`
`145`	`145`
`146`	`146`	`if (!CircuitState.IsHalfOpen)`
`147`	`147`	`{`
`148`		`- CircuitState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);`
	`148`	`+ MessagePumpCircuitState`
	`149`	`+ oldState = CircuitState,`
	`150`	`+ newState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);`
`149`	`151`
`150`		`- NotifyCircuitBreakerStateChangedSubscribers();`
	`152`	`+ CircuitState = newState;`
	`153`	`+ NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);`
`151`	`154`	`}`
`152`	`155`	`}`
`153`	`156`
`@@ -162,9 +165,12 @@ protected async Task WaitMessageIntervalDuringRecoveryAsync(CancellationToken ca`
`162`	`165`
`163`	`166`	`if (!CircuitState.IsHalfOpen)`
`164`	`167`	`{`
`165`		`- CircuitState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);`
	`168`	`+ MessagePumpCircuitState`
	`169`	`+ oldState = CircuitState,`
	`170`	`+ newState = CircuitState.TransitionTo(CircuitBreakerState.HalfOpen);`
`166`	`171`
`167`		`- NotifyCircuitBreakerStateChangedSubscribers();`
	`172`	`+ CircuitState = newState;`
	`173`	`+ NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);`
`168`	`174`	`}`
`169`	`175`	`}`
`170`	`176`
`@@ -176,9 +182,12 @@ internal void NotifyPauseReceiveMessages(MessagePumpCircuitBreakerOptions option`
`176`	`182`	`{`
`177`	`183`	`Logger.LogDebug("Circuit breaker caused message pump '{JobId}' to transition from a '{CurrentState}' an 'Open' state", JobId, CircuitState);`
`178`	`184`
`179`		`- CircuitState = CircuitState.TransitionTo(CircuitBreakerState.Open, options);`
	`185`	`+ MessagePumpCircuitState`
	`186`	`+ oldState = CircuitState,`
	`187`	`+ newState = CircuitState.TransitionTo(CircuitBreakerState.Open, options);`
`180`	`188`
`181`		`- NotifyCircuitBreakerStateChangedSubscribers();`
	`189`	`+ CircuitState = newState;`
	`190`	`+ NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);`
`182`	`191`	`}`
`183`	`192`
`184`	`193`	`/// <summary>`
`@@ -188,18 +197,21 @@ protected void NotifyResumeRetrievingMessages()`
`188`	`197`	`{`
`189`	`198`	`Logger.LogDebug("Circuit breaker caused message pump '{JobId}' to transition back from '{CurrentState}' to a 'Closed' state, retrieving messages is resumed", JobId, CircuitState);`
`190`	`199`
`191`		`- CircuitState = MessagePumpCircuitState.Closed;`
	`200`	`+ MessagePumpCircuitState`
	`201`	`+ oldState = CircuitState,`
	`202`	`+ newState = MessagePumpCircuitState.Closed;`
`192`	`203`
`193`		`- NotifyCircuitBreakerStateChangedSubscribers();`
	`204`	`+ CircuitState = newState;`
	`205`	`+ NotifyCircuitBreakerStateChangedSubscribers(oldState, newState);`
`194`	`206`	`}`
`195`	`207`
`196`		`- private void NotifyCircuitBreakerStateChangedSubscribers()`
	`208`	`+ private void NotifyCircuitBreakerStateChangedSubscribers(MessagePumpCircuitState oldState, MessagePumpCircuitState newState)`
`197`	`209`	`{`
`198`	`210`	`ICircuitBreakerEventHandler[] eventHandlers = GetEventHandlersForPump();`
`199`	`211`
`200`	`212`	`foreach (var handler in eventHandlers)`
`201`	`213`	`{`
`202`		`- Task.Run(() => handler.OnTransition(CircuitState));`
	`214`	`+ Task.Run(() => handler.OnTransition(new MessagePumpCircuitStateChangedEventArgs(JobId, oldState, newState)));`
`203`	`215`	`}`
`204`	`216`	`}`
`205`	`217`