Introduce CancellationToken to `WithCircuitBreaker` #7624

Arkatufus · 2025-05-06T22:23:48Z

Changes

Introduce CancellationToken to WithCircuitBreaker

The changes to CircuitBreaker is backward compatible, but the changes to Akka.Persistence is not, because these methods need to follow the new changes of the delegate function.

Warning

Breaking Change

Because the change was done to the delegate method fingerprint, it is impossible to make this change backward compatible.

All plugins that is based on Akka.Persistence will need to be updated to use these new Akka.Persistence internal APIs

Akka.Persistence.Sql
Akka.Persistence.MongoDb
Akka.Persistence.Azure
Akka.Persistence.Redis

Aaronontheweb · 2025-05-07T14:55:12Z

I checked the build warning counts to make sure that the obsoletions introduced in this PR don't create new warnings:

dev: 233
this PR: 233

So we're good to go there.

… into Harden-CircuitBreaker

Arkatufus

Self-review

Arkatufus · 2025-05-09T16:36:30Z

src/core/Akka.Persistence/Journal/AsyncRecovery.cs

        /// <returns>TBD</returns>
-        Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr);
+        Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr, CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:36:44Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

@@ -88,7 +90,7 @@ protected AsyncWriteJournal()
        public abstract Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> recoveryCallback);

        /// <inheritdoc/>
-        public abstract Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr);
+        public abstract Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr, CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:36:51Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

-        /// <returns>TBD</returns>
-        protected abstract Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerable<AtomicWrite> messages);
+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param> 
+        protected abstract Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerable<AtomicWrite> messages, CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:36:58Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

-        /// <returns>TBD</returns>
-        protected abstract Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr);
+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param> 
+        protected abstract Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr, CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:38:20Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

+                    await _breaker.WithCircuitBreaker((message, awj: this), (state, ct) =>
+                        state.awj.DeleteMessagesToAsync(state.message.PersistenceId, state.message.ToSequenceNr, ct));


Example of how CircuitBreaker will pass in its CancellationToken to the method it is protecting as a cancellation signal.

Arkatufus · 2025-05-09T16:42:09Z

src/core/Akka.Persistence/Snapshot/SnapshotStore.cs

+        protected abstract Task<SelectedSnapshot> LoadAsync(
+            string persistenceId, 
+            SnapshotSelectionCriteria criteria,
+            CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:42:17Z

src/core/Akka.Persistence/Snapshot/SnapshotStore.cs

+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param>
+        protected abstract Task SaveAsync(
+            SnapshotMetadata metadata,
+            object snapshot,
+            CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:42:30Z

src/core/Akka.Persistence/Snapshot/SnapshotStore.cs

-        /// <returns>TBD</returns>
-        protected abstract Task DeleteAsync(SnapshotMetadata metadata);
+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param>
+        protected abstract Task DeleteAsync(SnapshotMetadata metadata, CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:42:38Z

src/core/Akka.Persistence/Snapshot/SnapshotStore.cs

+        protected abstract Task DeleteAsync(
+            string persistenceId, 
+            SnapshotSelectionCriteria criteria,
+            CancellationToken cancellationToken);


[Breaking change] Akka.Persistence codes that are being wrapped inside a CircuitBreaker will now take a CancellationToken as a signal that the operation has been timed out by the CircuitBreaker.

Arkatufus · 2025-05-09T16:44:57Z

src/core/Akka/Pattern/CircuitBreaker.cs

+        /// <typeparam name="T">The <see cref="Type"/> returned by the protected function</typeparam>
+        /// <param name="body">Call needing protected</param>
+        /// <returns><see cref="Task"/> containing the call result</returns>
+        public Task<T> WithCircuitBreaker<T>(Func<CancellationToken, Task<T>> body) => CurrentState.Invoke(body);


All .WithCircuitBreaker() methods now accepts delegate function with a CancellationToken as one of its arguments, this CancellationToken will be used to pass in cancellation signal downstream into the protected delegate function if the CircuitBreaker throws.

…avior

Arkatufus · 2025-05-09T17:11:48Z

src/core/Akka/Util/Internal/AtomicState.cs

                CallSucceeds();
            }
            catch (Exception ex)
            {
+                cts.Cancel(); // Signal the protected delegate that operation has been canceled


Cancellation signal is sent to the underlying protected function here.

Aaronontheweb

LGTM

Aaronontheweb · 2025-05-09T19:12:30Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt

Aaronontheweb · 2025-05-09T19:12:51Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.DotNet.verified.txt

Unfortunate but necessary breaking change IMHO

Aaronontheweb · 2025-05-09T19:13:47Z

src/core/Akka.Persistence/Journal/AsyncRecovery.cs

        /// <returns>TBD</returns>
-        Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr);
+        Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr, CancellationToken cancellationToken);


Aaronontheweb · 2025-05-09T19:14:09Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

-                config.GetTimeSpan("circuit-breaker.reset-timeout", null));
+                config.GetInt("circuit-breaker.max-failures", 10),
+                config.GetTimeSpan("circuit-breaker.call-timeout", TimeSpan.FromSeconds(10)),
+                config.GetTimeSpan("circuit-breaker.reset-timeout", TimeSpan.FromSeconds(30)));


Seems high IMHO

But we can address that in a later issue if it is a problem - in fact I think 30s is the current default, isn't it?

Aaronontheweb · 2025-05-09T19:14:51Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

@@ -88,7 +90,7 @@ protected AsyncWriteJournal()
        public abstract Task ReplayMessagesAsync(IActorContext context, string persistenceId, long fromSequenceNr, long toSequenceNr, long max, Action<IPersistentRepresentation> recoveryCallback);

        /// <inheritdoc/>
-        public abstract Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr);
+        public abstract Task<long> ReadHighestSequenceNrAsync(string persistenceId, long fromSequenceNr, CancellationToken cancellationToken);


Aaronontheweb · 2025-05-09T19:14:55Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

-        /// <returns>TBD</returns>
-        protected abstract Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerable<AtomicWrite> messages);
+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param> 
+        protected abstract Task<IImmutableList<Exception>> WriteMessagesAsync(IEnumerable<AtomicWrite> messages, CancellationToken cancellationToken);


Aaronontheweb · 2025-05-09T19:15:02Z

src/core/Akka.Persistence/Journal/AsyncWriteJournal.cs

-        /// <returns>TBD</returns>
-        protected abstract Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr);
+        /// <param name="cancellationToken"><see cref="CancellationToken"/> used to signal cancelled snapshot operation</param> 
+        protected abstract Task DeleteMessagesToAsync(string persistenceId, long toSequenceNr, CancellationToken cancellationToken);


Aaronontheweb · 2025-05-09T19:15:45Z

src/core/Akka.Persistence/Snapshot/SnapshotStore.cs

@@ -59,155 +61,166 @@ private bool ReceiveSnapshotStore(object message)
            var senderPersistentActor = Sender; // Sender is PersistentActor
            var self = Self; //Self MUST BE CLOSED OVER here, or the code below will be subject to race conditions

-            if (message is LoadSnapshot loadSnapshot)
+            switch (message)


Aaronontheweb · 2025-05-09T19:41:00Z

src/core/Akka.Tests/Pattern/CircuitBreakerSpec.cs

Aaronontheweb · 2025-05-09T19:41:31Z

src/core/Akka/Pattern/CircuitBreaker.cs

Aaronontheweb · 2025-05-09T19:49:43Z

@Arkatufus just to validate that this design is going to fly with real plugins in the real world, can you do a fork using a local / testlab build of Akka.NET v1.5.42 and integrate it into something like Akka.Persistence.Sql or MongoDb?

We just want to validate that these changes can actually permeate down to the database drivers and do what we expect / hope (i.e. roll back a timed-out TXN in Akka.Persistence.Sql, for instance.)

No need for crazy unit or integration tests to validate this - that would be a gigantic waste of time. A quick "let's run a sample application with this change integrated into a real plugin and see what happens" would be fine.

Arkatufus · 2025-05-09T19:51:14Z

Will do

Aaronontheweb · 2025-05-12T16:44:40Z

Validated POC here akkadotnet/Akka.Persistence.Sql#535

Arkatufus added 4 commits May 7, 2025 05:22

Introduce CancellationToken to WithCircuitBreaker

1ca3cb1

Update API Approval list

46efbb3

Merge branch 'dev' into Harden-CircuitBreaker

422d2d0

Unify TaskCancelledException to TimeoutException (old behavior)

72589eb

Arkatufus marked this pull request as draft May 7, 2025 14:37

Arkatufus marked this pull request as ready for review May 7, 2025 16:07

Arkatufus and others added 13 commits May 8, 2025 02:31

Merge branch 'dev' into Harden-CircuitBreaker

9ea5975

Make Akka.Persistence API changes backward compatible

8238c0d

Merge branch 'Harden-CircuitBreaker' of github.com:Arkatufus/akka.net…

9eb97fc

… into Harden-CircuitBreaker

cleanup obsolete warnings

704731a

Merge branch 'dev' into Harden-CircuitBreaker

75c1660

Update API Approval list

fa4485a

Merge branch 'Harden-CircuitBreaker' of github.com:Arkatufus/akka.net…

4d8ef03

… into Harden-CircuitBreaker

Cleanup refactor warnings

ec19876

Merge branch 'dev' into Harden-CircuitBreaker

f154a82

Merge branch 'dev' into Harden-CircuitBreaker

d714290

Revert .ContinueWith() refactor

f0dfad8

Merge branch 'Harden-CircuitBreaker' of github.com:Arkatufus/akka.net…

7f1a0ce

… into Harden-CircuitBreaker

Code cleanup, remove stupid codes

98b351d

Arkatufus commented May 9, 2025

View reviewed changes

Code cleanup, make sure AtomicState.Callthrough() have the same beh…

7a5a20d

…avior

Arkatufus commented May 9, 2025

View reviewed changes

Update API Approval list

19e73df

Aaronontheweb added akka-persistence api-change labels May 9, 2025

Aaronontheweb approved these changes May 9, 2025

View reviewed changes

Aaronontheweb merged commit 0b91a04 into akkadotnet:dev May 12, 2025
11 checks passed

This was referenced May 12, 2025

CircuitBreaker timeout is decoupled from the underlying async operation #7617

Closed

Akka Persistence Throwing Unobserved Task Exception #7358

Closed

Arkatufus mentioned this pull request May 12, 2025

[EPIC] Update Akka.Persistence plugins to support new API #7632

Closed

3 tasks

Arkatufus mentioned this pull request May 21, 2025

Update RELEASE_NOTES.md for 1.5.42 release #7668

Merged

Arkatufus added this to the 1.5.42 milestone May 21, 2025

Arkatufus mentioned this pull request May 21, 2025

Fix: swallowed exceptions in timed out WithCircuitBreaker() #7362

Closed

This was referenced Jul 25, 2025

Bump Akka.Streams and 15 others Aaronontheweb/Alpakka#873

Closed

Bump Akka and 14 others petabridge/TurboMqtt#282

Closed

Bump Akka.Streams and 17 others Aaronontheweb/Alpakka#874

Closed

This was referenced Aug 15, 2025

Bump Akka.Streams from 1.5.40 to 1.5.47 Aaronontheweb/Alpakka#880

Closed

Bump Akka.Streams.TestKit from 1.5.40 to 1.5.47 Aaronontheweb/Alpakka#881

Closed

This was referenced Aug 25, 2025

Bump Akka.Streams from 1.5.40 to 1.5.48 Aaronontheweb/Alpakka#882

Open

Bump Akka.Streams.TestKit from 1.5.40 to 1.5.48 Aaronontheweb/Alpakka#883

Open

		await _breaker.WithCircuitBreaker((message, awj: this), (state, ct) =>
		state.awj.DeleteMessagesToAsync(state.message.PersistenceId, state.message.ToSequenceNr, ct));

Introduce CancellationToken to WithCircuitBreaker #7624

Introduce CancellationToken to WithCircuitBreaker #7624

Uh oh!

Conversation

Arkatufus commented May 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Breaking Change

Uh oh!

Aaronontheweb commented May 7, 2025

Uh oh!

Arkatufus left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb commented May 9, 2025

Uh oh!

Arkatufus commented May 9, 2025

Uh oh!

Aaronontheweb commented May 12, 2025

Uh oh!

Uh oh!

Uh oh!

Introduce CancellationToken to `WithCircuitBreaker` #7624

Introduce CancellationToken to `WithCircuitBreaker` #7624

Arkatufus commented May 6, 2025 •

edited

Loading