Ensure that BatchingSqlJournal will propagate sql connection opening failure

src/Akka.sln.DotSettings

@@ -19,6 +19,7 @@
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpKeepExistingMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpPlaceEmbeddedOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpRenamePlacementToArrangementMigration/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ECSharpUseContinuousIndentInsideBracesMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EAddAccessorOwnerDeclarationBracesMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002ECSharpPlaceAttributeOnSameLineMigration/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/Environment/SettingsMigration/IsMigratorApplied/=JetBrains_002EReSharper_002EPsi_002ECSharp_002ECodeStyle_002ESettingsUpgrade_002EMigrateBlankLinesAroundFieldToBlankLinesAroundProperty/@EntryIndexedValue">True</s:Boolean>

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/BatchingSqlJournal.cs

@@ -351,6 +351,18 @@ public abstract class BatchingSqlJournal<TConnection, TCommand> : WriteJournalBa
         where TCommand : DbCommand
     {
         #region internal classes
+
+        private sealed class ChunkExecutionFailure : IDeadLetterSuppression
+        {
+            public Exception Cause { get; }
+            public IJournalRequest[] Requests { get; }
+
+            public ChunkExecutionFailure(Exception cause, IJournalRequest[] requests)
+            {
+                Cause = cause;
+                Requests = requests;
+            }
+        }
 
         private sealed class BatchComplete
         {
@@ -658,10 +670,36 @@ protected sealed override bool Receive(object message)
             else if (message is Terminated) RemoveSubscriber(((Terminated)message).ActorRef);
             else if (message is GetCurrentPersistenceIds) InitializePersistenceIds();
             else if (message is CurrentPersistenceIds) SendCurrentPersistenceIds((CurrentPersistenceIds)message);
+            else if (message is ChunkExecutionFailure) FailChunkExecution((ChunkExecutionFailure)message);
             else return false;
             return true;
         }
 
+        private void FailChunkExecution(ChunkExecutionFailure message)
+        {
+            var cause = message.Cause;
+            Log.Error(cause, "Failed to execute chunk for {0} requests", message.Requests.Length);
+
+            foreach (var req in message.Requests)
+            {
+                switch (req)
+                {
+                    case WriteMessages write:
+                        write.PersistentActor.Tell(new WriteMessagesFailed(cause));
+                        break;
+                    case ReplayMessages replay:
+                        replay.PersistentActor.Tell(new ReplayMessagesFailure(cause));
+                        break;
+                    case DeleteMessagesTo delete:
+                        delete.PersistentActor.Tell(new DeleteMessagesFailure(cause, delete.ToSequenceNr));
+                        break;
+                    case ReplayTaggedMessages replayTagged:
+                        replayTagged.ReplyTo.Tell(new ReplayMessagesFailure(cause));
+                        break;
+                }
+            }
+        }
+
         private void SendCurrentPersistenceIds(CurrentPersistenceIds message)
         {
             foreach (var persistenceId in message.AllPersistenceIds)
@@ -837,7 +875,8 @@ private void TryProcess()
 
                 var chunk = DequeueChunk(_remainingOperations);
                 var context = Context;
-                _circuitBreaker.WithCircuitBreaker(() => ExecuteChunk(chunk, context)).PipeTo(Self);
+                _circuitBreaker.WithCircuitBreaker(() => ExecuteChunk(chunk, context))
+                    .PipeTo(Self, failure: ex => new ChunkExecutionFailure(ex, chunk.Requests));
             }
         }
 

src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Batching/BatchingSqliteJournalConnectionFailureSpec.cs

@@ -0,0 +1,40 @@
+//-----------------------------------------------------------------------
+// <copyright file="SqliteJournalConnectionFailureSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2018 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2018 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using Akka.Configuration;
+using Akka.Persistence.Sql.TestKit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.Sqlite.Tests.Batching
+{
+    public class BatchingSqliteJournalConnectionFailureSpec : SqlJournalConnectionFailureSpec
+    {
+        public BatchingSqliteJournalConnectionFailureSpec(ITestOutputHelper output)
+            : base(CreateSpecConfig(DefaultInvalidConnectionString), output)
+        {
+        }
+
+        private static Config CreateSpecConfig(string connectionString)
+        {
+            return ConfigurationFactory.ParseString(@"
+                akka.persistence {
+                    publish-plugin-commands = on
+                    journal {
+                        plugin = ""akka.persistence.journal.sqlite""
+                        sqlite {
+                            class = ""Akka.Persistence.Sqlite.Journal.BatchingSqliteJournal, Akka.Persistence.Sqlite""
+                            plugin-dispatcher = ""akka.actor.default-dispatcher""
+                            table-name = event_journal
+                            metadata-table-name = journal_metadata
+                            auto-initialize = on
+                            connection-string = """ + connectionString + @"""
+                        }
+                    }
+                }");
+        }
+    }
+}

src/core/Akka/Pattern/CircuitBreakerState.cs

@@ -37,7 +37,7 @@ public Open(CircuitBreaker breaker)
         /// <param name="body">N/A</param>
         /// <exception cref="OpenCircuitException">This exception is thrown automatically since the circuit is open.</exception>
         /// <returns>N/A</returns>
-        public override Task<T> Invoke<T>(Func<Task<T>> body)
+        public override async Task<T> Invoke<T>(Func<Task<T>> body)
         {
             throw new OpenCircuitException();
         }
@@ -48,7 +48,7 @@ public override Task<T> Invoke<T>(Func<Task<T>> body)
         /// <param name="body">N/A</param>
         /// <exception cref="OpenCircuitException">This exception is thrown automatically since the circuit is open.</exception>
         /// <returns>N/A</returns>
-        public override Task Invoke(Func<Task> body)
+        public override async Task Invoke(Func<Task> body)
         {
             throw new OpenCircuitException();
         }
@@ -106,13 +106,13 @@ public HalfOpen(CircuitBreaker breaker)
         /// <param name="body">Implementation of the call that needs protected</param>
         /// <exception cref="OpenCircuitException">TBD</exception>
         /// <returns><see cref="Task"/> containing result of protected call</returns>
-        public override Task<T> Invoke<T>(Func<Task<T>> body)
+        public override async Task<T> Invoke<T>(Func<Task<T>> body)
         {
             if (!_lock.CompareAndSet(true, false))
             {
                 throw new OpenCircuitException();
             }
-            return CallThrough(body);
+            return await CallThrough(body);
         }
 
         /// <summary>
@@ -122,13 +122,13 @@ public override Task<T> Invoke<T>(Func<Task<T>> body)
         /// <param name="body">Implementation of the call that needs protected</param>
         /// <exception cref="OpenCircuitException">TBD</exception>
         /// <returns><see cref="Task"/> containing result of protected call</returns>
-        public override Task Invoke(Func<Task> body)
+        public override async Task Invoke(Func<Task> body)
         {
             if (!_lock.CompareAndSet(true, false))
             {
                 throw new OpenCircuitException();
             }
-            return CallThrough(body);
+            await CallThrough(body);
         }
 
         /// <summary>