From 1e0941a38c76580c519b91cc6d225bbf205e3b71 Mon Sep 17 00:00:00 2001
From: atheate <a.theate@stariongroup.eu>
Date: Thu, 22 Jan 2026 17:01:36 +0100
Subject: [PATCH 1/5] Implemented safe subscription to also handle exception on
 calling action/function

---
 .../MessagingBackgroundServiceTestFixture.cs  | 105 +++++++++++++++++-
 .../MessagingBackgroundService.cs             |   6 +-
 .../ObservableExtensionsTestFixture.cs        |   8 +-
 Mercurio/Extensions/ObservableExtensions.cs   |  62 +++++++++--
 4 files changed, 165 insertions(+), 16 deletions(-)

diff --git a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs
index b63f57a..19c9320 100644
--- a/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs
+++ b/Mercurio.Hosting.Tests/MessagingBackgroundServiceTestFixture.cs
@@ -162,7 +162,7 @@ public async Task VerifyRpcServer()
             var rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 45, cancellationToken: cancellationTokenSource.Token);
             var taskCompletion = new TaskCompletionSource<bool>();
             rpcObservable.Subscribe(result => taskCompletion.SetResult(result));
-            await Task.Delay(50, cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
             
             await taskCompletion.Task;
             Assert.That(taskCompletion.Task.Result, Is.False);
@@ -170,7 +170,7 @@ public async Task VerifyRpcServer()
             rpcObservable = await client.SendRequestAsync(ConfiguredConnectionName, RpcServerQueueName, 44, cancellationToken: cancellationTokenSource.Token);
             var newTaskCompletion = new TaskCompletionSource<bool>();
             rpcObservable.Subscribe(result => newTaskCompletion.SetResult(result));
-            await Task.Delay(50, cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
             
             await taskCompletion.Task;
             Assert.That(newTaskCompletion.Task.Result, Is.True);
@@ -179,6 +179,56 @@ public async Task VerifyRpcServer()
             client.Dispose();
         }
 
+        [Test]
+        public async Task VerifyAutoRecoverySystem()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), this.configurationMock.Object);
+            _ = autoRecoveryBackground.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None);
+            
+            var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest");
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
+
+            Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(1));
+            
+            autoRecoveryBackground.PushMessage("123", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
+
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
+
+            Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3));
+            autoRecoveryBackground.Dispose();
+        }
+        
+        [Test]
+        public async Task VerifyAsyncAutoRecoverySystem()
+        {
+            using var cancellationTokenSource = new CancellationTokenSource();
+            var autoRecoveryBackground = new AutoRecoveryTestMessagingBackgroundService(this.serviceProvider, this.serviceProvider.GetRequiredService<ILogger<AutoRecoveryTestMessagingBackgroundService>>(), this.configurationMock.Object);
+            _ = autoRecoveryBackground.StartAsync(cancellationTokenSource.Token);
+            await Task.Delay(TimeSpan.FromMilliseconds(200), CancellationToken.None);
+            
+            var configuration = new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync");
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(100, cancellationTokenSource.Token);
+
+            Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(1));
+            
+            autoRecoveryBackground.PushMessage("123", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(50, cancellationTokenSource.Token);
+
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            autoRecoveryBackground.PushMessage("abc", configuration, cancellationToken: cancellationTokenSource.Token);
+            await Task.Delay(50, cancellationTokenSource.Token);
+
+            Assert.That(autoRecoveryBackground.ReceivedMessages, Has.Count.EqualTo(3));
+            autoRecoveryBackground.Dispose();
+        }
+
         private class InvalidInitializationBackgroundService : MessagingBackgroundService
         {
             /// <summary>
@@ -204,6 +254,57 @@ protected override Task InitializeAsync()
                 return Task.CompletedTask;
             }
         }
+        
+        public class AutoRecoveryTestMessagingBackgroundService: MessagingBackgroundService
+        {
+            /// <summary>
+            /// Stores all received message
+            /// </summary>
+            public readonly List<string> ReceivedMessages = [];
+            
+            /// <summary>
+            /// Initializes a new instance of the <see cref="MessagingBackgroundService" />
+            /// </summary>
+            /// <param name="serviceProvider">
+            /// The injected <see cref="IServiceProvider" /> that allow to resolve
+            /// <see cref="IMessageClientService" /> instance, even if not registered as scope
+            /// </param>
+            /// <param name="logger"></param>
+            /// <param name="configuration"></param>
+            public AutoRecoveryTestMessagingBackgroundService(IServiceProvider serviceProvider, ILogger<AutoRecoveryTestMessagingBackgroundService> logger, IConfiguration configuration) : base(serviceProvider, logger, configuration)
+            {
+            }
+
+            /// <summary>
+            /// Initializes this service (e.g. to set the <see cref="MessagingBackgroundService.ConnectionName" /> and register subscriptions
+            /// collection
+            /// </summary>
+            /// <returns>An awaitable <see cref="Task" /></returns>
+            protected override async Task InitializeAsync()
+            {
+                this.ConnectionName = ConfiguredConnectionName;
+                await this.RegisterListener(() => this.MessageClientService.ListenAsync<string>(this.ConnectionName, new FanoutExchangeConfiguration("AutoRecoveryBackgroundTest")), this.HandleMessage);
+                await this.RegisterAsyncListener(() => this.MessageClientService.ListenAsync<string>(this.ConnectionName, new FanoutExchangeConfiguration("AutoRecoveryBackgroundTestAsync")), this.HandleMessageAsync);
+            }
+
+            private Task HandleMessageAsync(string arg)
+            {
+                this.HandleMessage(arg);
+                return Task.CompletedTask;
+            }
+
+            private void HandleMessage(string obj)
+            {
+                if (int.TryParse(obj, out _))
+                {
+                    throw new InvalidOperationException("I want to throw something");
+                }
+                else
+                {
+                    this.ReceivedMessages.Add(obj);
+                }
+            }
+        }
 
         public class TestMessagingBackgroundService: MessagingBackgroundService
         {
diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs
index d9b39a3..1b2b8fa 100644
--- a/Mercurio.Hosting/MessagingBackgroundService.cs
+++ b/Mercurio.Hosting/MessagingBackgroundService.cs
@@ -232,7 +232,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                     this.Logger.LogError(exception, "Error occured while processing a message");
                 }
 
-                await Task.Delay(TimeSpan.FromMilliseconds(10), stoppingToken);
+                await Task.Delay(TimeSpan.FromMilliseconds(5), stoppingToken);
             }
 
             this.IsHealthy = false;
@@ -262,7 +262,7 @@ protected async Task RegisterListener<TMessage>(Func<Task<IObservable<TMessage>>
             var key = Guid.NewGuid();
 
             var observable = await observableFunction();
-            var subscription = observable.Subscribe(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
+            var subscription = observable.SubscribeSafe(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
             this.subscriptions[key] = subscription;
         }
 
@@ -285,7 +285,7 @@ protected async Task RegisterAsyncListener<TMessage>(Func<Task<IObservable<TMess
             var key = Guid.NewGuid();
 
             var observable = await observableFunction();
-            var subscription = observable.SubscribeAsync(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
+            var subscription = observable.SubscribeSafeAsync(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
             this.subscriptions[key] = subscription;
         }
 
diff --git a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
index 670125c..52a4f04 100644
--- a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
+++ b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
@@ -36,10 +36,10 @@ public void VerifyNullArgumentsThrow()
             using (Assert.EnterMultipleScope())
             {
                 IObservable<int> source = null;
-                Assert.That(() => source.SubscribeAsync(_ => Task.CompletedTask), Throws.ArgumentNullException);
+                Assert.That(() => source.SubscribeSafeAsync(_ => Task.CompletedTask), Throws.ArgumentNullException);
 
                 var observable = new Subject<int>();
-                Assert.That(() => observable.SubscribeAsync(null), Throws.ArgumentNullException);
+                Assert.That(() => observable.SubscribeSafeAsync(null), Throws.ArgumentNullException);
             }
         }
 
@@ -50,7 +50,7 @@ public async Task VerifySubscription()
             var completed = false;
             var subject = new Subject<int>();
 
-            using var subscription = subject.SubscribeAsync(x =>
+            using var subscription = subject.SubscribeSafeAsync(x =>
             {
                 results.Add(x);
                 return Task.CompletedTask;
@@ -75,7 +75,7 @@ public async Task VerifyErrorIsHandled()
             var subject = new Subject<int>();
             Exception captured = null;
 
-            using var subscription = subject.SubscribeAsync(_ => Task.FromException(new InvalidOperationException("boom")), ex => captured = ex);
+            using var subscription = subject.SubscribeSafeAsync(_ => Task.FromException(new InvalidOperationException("boom")), ex => captured = ex);
 
             subject.OnNext(1);
 
diff --git a/Mercurio/Extensions/ObservableExtensions.cs b/Mercurio/Extensions/ObservableExtensions.cs
index a9fa0bc..7dfd7a9 100644
--- a/Mercurio/Extensions/ObservableExtensions.cs
+++ b/Mercurio/Extensions/ObservableExtensions.cs
@@ -1,7 +1,7 @@
 // -------------------------------------------------------------------------------------------------
 //  <copyright file="ObservableExtensions.cs" company="Starion Group S.A.">
 // 
-//    Copyright 2025 Starion Group S.A.
+//    Copyright 2026 Starion Group S.A.
 // 
 //    Licensed under the Apache License, Version 2.0 (the "License");
 //    you may not use this file except in compliance with the License.
@@ -28,15 +28,17 @@ namespace Mercurio.Extensions
     public static class ObservableExtensions
     {
         /// <summary>
-        /// Subscribe to an <see cref="IObservable{T}" /> with async capabilities
+        /// Safely subscribe to an <see cref="IObservable{T}" /> with async capabilities
         /// </summary>
         /// <typeparam name="T">An object</typeparam>
         /// <param name="source">The source <see cref="IObservable{T}" /></param>
         /// <param name="onNextAsync">The <see cref="Func{TResult,UResult}" /> to call</param>
-        /// <param name="onError">An optional <see cref="Action{T}"/> to handle exception</param>
-        /// <param name="onCompleted">An optional <see cref="Action"/> to handle completed action</param>
+        /// <param name="onError">An optional <see cref="Action{T}" /> to handle exception</param>
+        /// <param name="onCompleted">An optional <see cref="Action" /> to handle completed action</param>
+        /// <exception cref="ArgumentNullException">If one of the provided <paramref name="source"/> or <paramref name="onNextAsync"/>
+        /// is null</exception>
         /// <returns>The created <see cref="IDisposable" /></returns>
-        public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T, Task> onNextAsync, Action<Exception> onError = null,
+        public static IDisposable SubscribeSafeAsync<T>(this IObservable<T> source, Func<T, Task> onNextAsync, Action<Exception> onError = null,
             Action onCompleted = null)
         {
             if (source == null)
@@ -49,9 +51,55 @@ public static IDisposable SubscribeAsync<T>(this IObservable<T> source, Func<T,
                 throw new ArgumentNullException(nameof(onNextAsync));
             }
 
-            return source.Select(x => Observable.FromAsync(() => onNextAsync(x))).Concat()
+            return source.Select(x => Observable.FromAsync(() =>
+                {
+                    try
+                    {
+                        return onNextAsync(x);
+                    }
+                    catch (Exception e)
+                    {
+                        onError?.Invoke(e);
+                        return Task.CompletedTask;
+                    }
+                })).Concat()
                 .Subscribe(_ => { }, onError ?? (_ => { }), onCompleted ?? (() => { }));
         }
+
+        /// <summary>
+        /// Safely subscribe to an <see cref="IObservable{T}" /> 
+        /// </summary>
+        /// <typeparam name="T">An object</typeparam>
+        /// <param name="source">The source <see cref="IObservable{T}" /></param>
+        /// <param name="onNext">The <see cref="Action{T}" /> to call</param>
+        /// <param name="onError">An optional <see cref="Action{T}" /> to handle exception</param>
+        /// <param name="onCompleted">An optional <see cref="Action" /> to handle completed action</param>
+        /// <exception cref="ArgumentNullException">If one of the provided <paramref name="source"/> or <paramref name="onNext"/>
+        /// is null</exception>
+        /// <returns>The created <see cref="IDisposable" /></returns>
+        public static IDisposable SubscribeSafe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError = null, Action onCompleted = null)
+        {
+            if (source == null)
+            {
+                throw new ArgumentNullException(nameof(source));
+            }
+
+            if (onNext == null)
+            {
+                throw new ArgumentNullException(nameof(onNext));
+            }
+
+            return source.Subscribe(x =>
+            {
+                try
+                {
+                    onNext(x);
+                }
+                catch (Exception e)
+                {
+                    onError?.Invoke(e);
+                }
+            }, onError ?? (_ => { }), onCompleted ?? (() => { }));
+        }
     }
 }
-    
\ No newline at end of file

From 106f0cdb342d88e568b5d288aedfabaa0dfbdd47 Mon Sep 17 00:00:00 2001
From: atheate <a.theate@stariongroup.eu>
Date: Thu, 22 Jan 2026 17:07:15 +0100
Subject: [PATCH 2/5] Moved from sonar.login to sonar.token

---
 .github/workflows/CodeQuality.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/CodeQuality.yml b/.github/workflows/CodeQuality.yml
index 4242334..a97166b 100644
--- a/.github/workflows/CodeQuality.yml
+++ b/.github/workflows/CodeQuality.yml
@@ -37,7 +37,7 @@ jobs:
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
         run: |
           dotnet tool install --global dotnet-sonarscanner
-          dotnet sonarscanner begin /k:"STARIONGROUP_Mercurio" /o:"stariongroup" /d:sonar.login="$SONAR_TOKEN" /d:sonar.host.url="https://sonarcloud.io" /d:sonar.cs.opencover.reportsPaths="./CoverageResults/coverage.opencover.xml"
+          dotnet sonarscanner begin /k:"STARIONGROUP_Mercurio" /o:"stariongroup" /d:sonar.token="$SONAR_TOKEN" /d:sonar.host.url="https://sonarcloud.io" /d:sonar.cs.opencover.reportsPaths="./CoverageResults/coverage.opencover.xml"
 
       - name: Build
         run: dotnet build --no-restore /p:ContinuousIntegrationBuild=true
@@ -51,5 +51,5 @@ jobs:
       - name: Sonarqube end
         env:
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-        run: dotnet sonarscanner end /d:sonar.login="$SONAR_TOKEN"
+        run: dotnet sonarscanner end /d:sonar.token="$SONAR_TOKEN"
         
\ No newline at end of file

From 15aed7563910c720743648c84cd6140452d7f44b Mon Sep 17 00:00:00 2001
From: atheate <a.theate@stariongroup.eu>
Date: Fri, 23 Jan 2026 08:45:30 +0100
Subject: [PATCH 3/5] Upgrade deps and does not recover observable if it's the
 action that throws exception

---
 .../Mercurio.Core.Tests.csproj                |  2 +-
 .../RabbitMqContainerSetupFixture.cs          |  3 +-
 .../Mercurio.Hosting.Tests.csproj             |  6 ++--
 Mercurio.Hosting/Mercurio.Hosting.csproj      |  6 ++--
 .../MessagingBackgroundService.cs             |  4 +--
 .../ObservableExtensionsTestFixture.cs        |  2 +-
 Mercurio.Tests/Mercurio.Tests.csproj          | 28 +++++++++----------
 Mercurio/Extensions/ObservableExtensions.cs   | 18 ++++++------
 Mercurio/Mercurio.csproj                      | 10 +++----
 9 files changed, 40 insertions(+), 39 deletions(-)

diff --git a/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj b/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj
index 89c576b..becf5fc 100644
--- a/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj
+++ b/Mercurio.Core.Tests/Mercurio.Core.Tests.csproj
@@ -11,7 +11,7 @@
 
     <ItemGroup>
         <PackageReference Include="NUnit" Version="4.4.0" />
-        <PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
+        <PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
     </ItemGroup>
 
 </Project>
diff --git a/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs b/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs
index c533e97..10206b3 100644
--- a/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs
+++ b/Mercurio.Core.Tests/RabbitMqContainerSetupFixture.cs
@@ -32,8 +32,7 @@ public class RabbitMqContainerSetupFixture
         [OneTimeSetUp]
         public async Task Setup()
         {
-            RabbitMqContainer = new RabbitMqBuilder()
-                .WithImage("rabbitmq:4.2.0-alpine")
+            RabbitMqContainer = new RabbitMqBuilder("rabbitmq:4.2.0-alpine")
                 .WithUsername("guest")
                 .WithPassword("guest")
                 .Build();
diff --git a/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj b/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj
index 81791be..162a775 100644
--- a/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj
+++ b/Mercurio.Hosting.Tests/Mercurio.Hosting.Tests.csproj
@@ -21,13 +21,13 @@
         <PrivateAssets>all</PrivateAssets>
         <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="NUnit3TestAdapter" Version="5.2.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
+    <PackageReference Include="NUnit3TestAdapter" Version="6.1.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.2" />
     <PackageReference Include="coverlet.msbuild" Version="6.0.4">
         <PrivateAssets>all</PrivateAssets>
         <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-        <PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
+        <PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
     </ItemGroup>
 
     <ItemGroup>
diff --git a/Mercurio.Hosting/Mercurio.Hosting.csproj b/Mercurio.Hosting/Mercurio.Hosting.csproj
index dbd0267..ede1161 100644
--- a/Mercurio.Hosting/Mercurio.Hosting.csproj
+++ b/Mercurio.Hosting/Mercurio.Hosting.csproj
@@ -43,9 +43,9 @@
     </ItemGroup>
 
     <ItemGroup>
-      <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions" Version="10.0.0" />
-      <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
-      <PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.3" PrivateAssets="all" />
+      <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks.Abstractions" Version="10.0.2" />
+      <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.2" />
+      <PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.5" PrivateAssets="all" />
     </ItemGroup>
 
 </Project>
diff --git a/Mercurio.Hosting/MessagingBackgroundService.cs b/Mercurio.Hosting/MessagingBackgroundService.cs
index 1b2b8fa..00effa7 100644
--- a/Mercurio.Hosting/MessagingBackgroundService.cs
+++ b/Mercurio.Hosting/MessagingBackgroundService.cs
@@ -262,7 +262,7 @@ protected async Task RegisterListener<TMessage>(Func<Task<IObservable<TMessage>>
             var key = Guid.NewGuid();
 
             var observable = await observableFunction();
-            var subscription = observable.SubscribeSafe(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
+            var subscription = observable.SubscribeSafe(onReceive,onNextError:onError, onObservableError: x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted: onCompleted);
             this.subscriptions[key] = subscription;
         }
 
@@ -285,7 +285,7 @@ protected async Task RegisterAsyncListener<TMessage>(Func<Task<IObservable<TMess
             var key = Guid.NewGuid();
 
             var observable = await observableFunction();
-            var subscription = observable.SubscribeSafeAsync(onReceive, x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted);
+            var subscription = observable.SubscribeSafeAsync(onReceive, onNextError: onError,onObservableError: x => _ = this.HandleErrorAndRecover(x, observableFunction, onReceive, onError, onCompleted, key), onCompleted: onCompleted);
             this.subscriptions[key] = subscription;
         }
 
diff --git a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
index 52a4f04..185e4bd 100644
--- a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
+++ b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
@@ -75,7 +75,7 @@ public async Task VerifyErrorIsHandled()
             var subject = new Subject<int>();
             Exception captured = null;
 
-            using var subscription = subject.SubscribeSafeAsync(_ => Task.FromException(new InvalidOperationException("boom")), ex => captured = ex);
+            using var subscription = subject.SubscribeSafeAsync(_ => Task.FromException(new InvalidOperationException("boom")), onObservableError: ex => captured = ex);
 
             subject.OnNext(1);
 
diff --git a/Mercurio.Tests/Mercurio.Tests.csproj b/Mercurio.Tests/Mercurio.Tests.csproj
index 104b745..8803245 100644
--- a/Mercurio.Tests/Mercurio.Tests.csproj
+++ b/Mercurio.Tests/Mercurio.Tests.csproj
@@ -13,9 +13,9 @@
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="Microsoft.Extensions.Configuration" Version="10.0.0" />
-        <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.0" />
-        <PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.0" />
+        <PackageReference Include="Microsoft.Extensions.Configuration" Version="10.0.2" />
+        <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.2" />
+        <PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.2" />
         <PackageReference Include="Moq" Version="4.20.72" />
         <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.0.1" />
         <PackageReference Include="NUnit" Version="4.4.0" />
@@ -23,21 +23,21 @@
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="NUnit3TestAdapter" Version="5.2.0" />
-        <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
+        <PackageReference Include="NUnit3TestAdapter" Version="6.1.0" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.2" />
         <PackageReference Include="coverlet.msbuild" Version="6.0.4">
             <PrivateAssets>all</PrivateAssets>
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
-        <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
-        <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
-        <PackageReference Include="OpenTelemetry" Version="1.14.0" />
-        <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.14.0" />
-        <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.14.0" />
-        <PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.14.0" />
-        <PackageReference Include="OpenTelemetry.Instrumentation.Http" Version="1.14.0" />
-        <PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.14.0" />
-        <PackageReference Include="Testcontainers.RabbitMq" Version="4.8.1" />
+        <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.2" />
+        <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.2" />
+        <PackageReference Include="OpenTelemetry" Version="1.15.0" />
+        <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.15.0" />
+        <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.15.0" />
+        <PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.15.0" />
+        <PackageReference Include="OpenTelemetry.Instrumentation.Http" Version="1.15.0" />
+        <PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.15.0" />
+        <PackageReference Include="Testcontainers.RabbitMq" Version="4.10.0" />
     </ItemGroup>
 
     <ItemGroup>
diff --git a/Mercurio/Extensions/ObservableExtensions.cs b/Mercurio/Extensions/ObservableExtensions.cs
index 7dfd7a9..4425767 100644
--- a/Mercurio/Extensions/ObservableExtensions.cs
+++ b/Mercurio/Extensions/ObservableExtensions.cs
@@ -33,12 +33,13 @@ public static class ObservableExtensions
         /// <typeparam name="T">An object</typeparam>
         /// <param name="source">The source <see cref="IObservable{T}" /></param>
         /// <param name="onNextAsync">The <see cref="Func{TResult,UResult}" /> to call</param>
-        /// <param name="onError">An optional <see cref="Action{T}" /> to handle exception</param>
+        /// <param name="onNextError">An optional <see cref="Action{T}" /> to handle exception thrown by the <paramref name="onNextAsync"/></param>
+        /// <param name="onObservableError">An optional <see cref="Action{T}" /> to handle exception thrown by the <paramref name="source"/></param>
         /// <param name="onCompleted">An optional <see cref="Action" /> to handle completed action</param>
         /// <exception cref="ArgumentNullException">If one of the provided <paramref name="source"/> or <paramref name="onNextAsync"/>
         /// is null</exception>
         /// <returns>The created <see cref="IDisposable" /></returns>
-        public static IDisposable SubscribeSafeAsync<T>(this IObservable<T> source, Func<T, Task> onNextAsync, Action<Exception> onError = null,
+        public static IDisposable SubscribeSafeAsync<T>(this IObservable<T> source, Func<T, Task> onNextAsync, Action<Exception> onNextError = null, Action<Exception> onObservableError = null,
             Action onCompleted = null)
         {
             if (source == null)
@@ -59,11 +60,11 @@ public static IDisposable SubscribeSafeAsync<T>(this IObservable<T> source, Func
                     }
                     catch (Exception e)
                     {
-                        onError?.Invoke(e);
+                        onNextError?.Invoke(e);
                         return Task.CompletedTask;
                     }
                 })).Concat()
-                .Subscribe(_ => { }, onError ?? (_ => { }), onCompleted ?? (() => { }));
+                .Subscribe(_ => { }, onObservableError ?? (_ => { }), onCompleted ?? (() => { }));
         }
 
         /// <summary>
@@ -72,12 +73,13 @@ public static IDisposable SubscribeSafeAsync<T>(this IObservable<T> source, Func
         /// <typeparam name="T">An object</typeparam>
         /// <param name="source">The source <see cref="IObservable{T}" /></param>
         /// <param name="onNext">The <see cref="Action{T}" /> to call</param>
-        /// <param name="onError">An optional <see cref="Action{T}" /> to handle exception</param>
+        /// <param name="onNextError">An optional <see cref="Action{T}" /> to handle exception thrown by the <paramref name="onNext"/></param>
+        /// <param name="onObservableError">An optional <see cref="Action{T}" /> to handle exception thrown by the <paramref name="source"/></param>
         /// <param name="onCompleted">An optional <see cref="Action" /> to handle completed action</param>
         /// <exception cref="ArgumentNullException">If one of the provided <paramref name="source"/> or <paramref name="onNext"/>
         /// is null</exception>
         /// <returns>The created <see cref="IDisposable" /></returns>
-        public static IDisposable SubscribeSafe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onError = null, Action onCompleted = null)
+        public static IDisposable SubscribeSafe<T>(this IObservable<T> source, Action<T> onNext, Action<Exception> onNextError = null, Action<Exception> onObservableError = null, Action onCompleted = null)
         {
             if (source == null)
             {
@@ -97,9 +99,9 @@ public static IDisposable SubscribeSafe<T>(this IObservable<T> source, Action<T>
                 }
                 catch (Exception e)
                 {
-                    onError?.Invoke(e);
+                    onNextError?.Invoke(e);
                 }
-            }, onError ?? (_ => { }), onCompleted ?? (() => { }));
+            }, onObservableError ?? (_ => { }), onCompleted ?? (() => { }));
         }
     }
 }
diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj
index 1e7337e..2179a1b 100644
--- a/Mercurio/Mercurio.csproj
+++ b/Mercurio/Mercurio.csproj
@@ -34,13 +34,13 @@
 
     <ItemGroup>
         <PackageReference Include="CommunityToolkit.HighPerformance" Version="8.4.0" />
-        <PackageReference Include="Microsoft.AspNetCore.Hosting.Abstractions" Version="2.3.0" />
-        <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
-        <PackageReference Include="Polly" Version="8.6.4" />
+        <PackageReference Include="Microsoft.AspNetCore.Hosting.Abstractions" Version="2.3.9" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.2" />
+        <PackageReference Include="Polly" Version="8.6.5" />
         <PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
         <PackageReference Include="System.Reactive" Version="6.1.0" />
-        <PackageReference Include="System.Text.Json" Version="10.0.0" />
-        <PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.3" PrivateAssets="all" />
+        <PackageReference Include="System.Text.Json" Version="10.0.2" />
+        <PackageReference Include="Microsoft.Sbom.Targets" Version="4.1.5" PrivateAssets="all" />
     </ItemGroup>
 
     <ItemGroup>

From d06c3db94666a39a9bcc3bc76546d3420f1fed8d Mon Sep 17 00:00:00 2001
From: atheate <a.theate@stariongroup.eu>
Date: Fri, 23 Jan 2026 08:57:08 +0100
Subject: [PATCH 4/5] Code coverage

---
 Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
index 185e4bd..43e9538 100644
--- a/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
+++ b/Mercurio.Tests/Extensions/ObservableExtensionsTestFixture.cs
@@ -37,9 +37,11 @@ public void VerifyNullArgumentsThrow()
             {
                 IObservable<int> source = null;
                 Assert.That(() => source.SubscribeSafeAsync(_ => Task.CompletedTask), Throws.ArgumentNullException);
+                Assert.That(() => source.SubscribeSafe(_ => { }), Throws.ArgumentNullException);
 
                 var observable = new Subject<int>();
                 Assert.That(() => observable.SubscribeSafeAsync(null), Throws.ArgumentNullException);
+                Assert.That(() => observable.SubscribeSafe(onNext: null), Throws.ArgumentNullException);
             }
         }
 

From 1c6557bb04b62dfedb71eb5ed95fd1da5ba24072 Mon Sep 17 00:00:00 2001
From: atheate <a.theate@stariongroup.eu>
Date: Fri, 23 Jan 2026 09:01:27 +0100
Subject: [PATCH 5/5] Version bump

---
 Mercurio.Hosting/Mercurio.Hosting.csproj | 4 ++--
 Mercurio/Mercurio.csproj                 | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/Mercurio.Hosting/Mercurio.Hosting.csproj b/Mercurio.Hosting/Mercurio.Hosting.csproj
index ede1161..3e6571c 100644
--- a/Mercurio.Hosting/Mercurio.Hosting.csproj
+++ b/Mercurio.Hosting/Mercurio.Hosting.csproj
@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>disable</Nullable>
         <TargetFrameworks>net10.0;net9.0;net8.0;netstandard2.1</TargetFrameworks>
-        <Version>2.0.0</Version>
+        <Version>2.0.1</Version>
         <LangVersion>12.0</LangVersion>
         <Company>Starion Group S.A.</Company>
         <Copyright>Copyright © Starion Group S.A.</Copyright>
@@ -28,7 +28,7 @@
         <GenerateSBOM>true</GenerateSBOM>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <PackageReleaseNotes>
-            [BUMP] Bump to Net10.0
+            [FIX] Fix Bug with thrown exception during onNext Action/Function
         </PackageReleaseNotes>
     </PropertyGroup>
 
diff --git a/Mercurio/Mercurio.csproj b/Mercurio/Mercurio.csproj
index 2179a1b..33bbf5a 100644
--- a/Mercurio/Mercurio.csproj
+++ b/Mercurio/Mercurio.csproj
@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>disable</Nullable>
         <TargetFrameworks>net10.0;net9.0;net8.0;netstandard2.1</TargetFrameworks>
-        <Version>2.0.0</Version>
+        <Version>2.0.1</Version>
         <LangVersion>12.0</LangVersion>
         <Company>Starion Group S.A.</Company>
         <Copyright>Copyright © Starion Group S.A.</Copyright>
@@ -28,7 +28,7 @@
         <GenerateSBOM>true</GenerateSBOM>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
         <PackageReleaseNotes>
-            [BUMP] Bump to Net10.0
+            [FIX] Fix Bug with thrown exception during onNext Action/Function
         </PackageReleaseNotes>
     </PropertyGroup>