Fix sharding recovery error and WithTransport serialization

src/contrib/cluster/Akka.Cluster.Sharding.Tests/SupervisionSpec.cs

@@ -12,6 +12,7 @@
 using Akka.Event;
 using Akka.Pattern;
 using Xunit;
+using Xunit.Abstractions;
 
 namespace Akka.Cluster.Sharding.Tests
 {
@@ -93,7 +94,7 @@ protected override void OnReceive(object message)
         private readonly ExtractShardId _extractShard = message =>
             message is Msg msg ? (msg.Id % 2).ToString(CultureInfo.InvariantCulture) : null;
 
-        public SupervisionSpec() : base(GetConfig())
+        public SupervisionSpec(ITestOutputHelper output) : base(GetConfig(), output: output)
         { }
 
         public static Config GetConfig()

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

@@ -1099,8 +1099,7 @@ private async Task HandleWriteMessages(WriteMessages req, TCommand command)
 
                 foreach (var envelope in req.Messages)
                 {
-                    var write = envelope as AtomicWrite;
-                    if (write != null)
+                    if (envelope is AtomicWrite write)
                     {
                         var writes = (IImmutableList<IPersistentRepresentation>)write.Payload;
                         foreach (var unadapted in writes)
@@ -1111,9 +1110,8 @@ private async Task HandleWriteMessages(WriteMessages req, TCommand command)
                                 tagBuilder.Clear();
 
                                 var persistent = AdaptToJournal(unadapted);
-                                if (persistent.Payload is Tagged)
+                                if (persistent.Payload is Tagged tagged)
                                 {
-                                    var tagged = (Tagged)persistent.Payload;
                                     if (tagged.Tags.Count != 0)
                                     {
                                         tagBuilder.Append(';');
@@ -1205,29 +1203,35 @@ protected virtual void WriteEvent(TCommand command, IPersistentRepresentation pe
             var payloadType = persistent.Payload.GetType();
             var serializer = _serialization.FindSerializerForType(payloadType, Setup.DefaultSerializer);
 
-            string manifest = "";
-            if (serializer is SerializerWithStringManifest)
+            // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+            Akka.Serialization.Serialization.WithTransport(_serialization.System, () =>
             {
-                manifest = ((SerializerWithStringManifest)serializer).Manifest(persistent.Payload);
-            }
-            else
-            {
-                if (serializer.IncludeManifest)
+                string manifest = "";
+                if (serializer is SerializerWithStringManifest stringManifest)
                 {
-                    manifest = persistent.Payload.GetType().TypeQualifiedName();
+                    manifest = stringManifest.Manifest(persistent.Payload);
                 }
-            }
+                else
+                {
+                    if (serializer.IncludeManifest)
+                    {
+                        manifest = persistent.Payload.GetType().TypeQualifiedName();
+                    }
+                }
+
+                var binary = serializer.ToBinary(persistent.Payload);
 
-            var binary = serializer.ToBinary(persistent.Payload);
+                AddParameter(command, "@PersistenceId", DbType.String, persistent.PersistenceId);
+                AddParameter(command, "@SequenceNr", DbType.Int64, persistent.SequenceNr);
+                AddParameter(command, "@Timestamp", DbType.Int64, 0L);
+                AddParameter(command, "@IsDeleted", DbType.Boolean, false);
+                AddParameter(command, "@Manifest", DbType.String, manifest);
+                AddParameter(command, "@Payload", DbType.Binary, binary);
+                AddParameter(command, "@Tag", DbType.String, tags);
+                AddParameter(command, "@SerializerId", DbType.Int32, serializer.Identifier);
 
-            AddParameter(command, "@PersistenceId", DbType.String, persistent.PersistenceId);
-            AddParameter(command, "@SequenceNr", DbType.Int64, persistent.SequenceNr);
-            AddParameter(command, "@Timestamp", DbType.Int64, 0L);
-            AddParameter(command, "@IsDeleted", DbType.Boolean, false);
-            AddParameter(command, "@Manifest", DbType.String, manifest);
-            AddParameter(command, "@Payload", DbType.Binary, binary);
-            AddParameter(command, "@Tag", DbType.String, tags);
-            AddParameter(command, "@SerializerId", DbType.Int32, serializer.Identifier);
+                return manifest;
+            });
         }
 
         /// <summary>

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

@@ -690,20 +690,26 @@ protected virtual void WriteEvent(DbCommand command, IPersistentRepresentation e
             var payloadType = e.Payload.GetType();
             var serializer = Serialization.FindSerializerForType(payloadType, Configuration.DefaultSerializer);
 
+            // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
             string manifest = "";
-            if (serializer is SerializerWithStringManifest)
+            var binary = Akka.Serialization.Serialization.WithTransport(Serialization.System, () =>
             {
-                manifest = ((SerializerWithStringManifest)serializer).Manifest(e.Payload);
-            }
-            else
-            {
-                if (serializer.IncludeManifest)
+
+                if (serializer is SerializerWithStringManifest stringManifest)
                 {
-                    manifest = e.Payload.GetType().TypeQualifiedName();
+                    manifest = stringManifest.Manifest(e.Payload);
+                }
+                else
+                {
+                    if (serializer.IncludeManifest)
+                    {
+                        manifest = e.Payload.GetType().TypeQualifiedName();
+                    }
                 }
-            }
 
-            var binary = serializer.ToBinary(e.Payload);
+                return serializer.ToBinary(e.Payload);
+            });
+
 
             AddParameter(command, "@PersistenceId", DbType.String, e.PersistenceId);
             AddParameter(command, "@SequenceNr", DbType.Int64, e.SequenceNr);
@@ -746,11 +752,13 @@ protected virtual IPersistentRepresentation ReadEvent(DbDataReader reader)
                 // Support old writes that did not set the serializer id
                 var type = Type.GetType(manifest, true);
                 var deserializer = Serialization.FindSerializerForType(type, Configuration.DefaultSerializer);
-                deserialized = deserializer.FromBinary((byte[])payload, type);
+                // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+                deserialized = Akka.Serialization.Serialization.WithTransport(Serialization.System, () => deserializer.FromBinary((byte[])payload, type) );
             }
             else
             {
                 var serializerId = reader.GetInt32(SerializerIdIndex);
+                // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
                 deserialized = Serialization.Deserialize((byte[])payload, serializerId, manifest);
             }
 

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/QueryExecutor.cs

@@ -337,8 +337,8 @@ protected virtual void SetPayloadParameter(object snapshot, DbCommand command)
         {
             var snapshotType = snapshot.GetType();
             var serializer = Serialization.FindSerializerForType(snapshotType, Configuration.DefaultSerializer);
-
-            var binary = serializer.ToBinary(snapshot);
+            // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+            var binary = Akka.Serialization.Serialization.WithTransport(Serialization.System, () => serializer.ToBinary(snapshot));
             AddParameter(command, "@Payload", DbType.Binary, binary);
         }
 
@@ -585,8 +585,9 @@ protected object GetSnapshot(DbDataReader reader)
             if (reader.IsDBNull(5))
             {
                 var type = Type.GetType(manifest, true);
+                // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
                 var serializer = Serialization.FindSerializerForType(type, Configuration.DefaultSerializer);
-                obj = serializer.FromBinary(binary, type);
+                obj = Akka.Serialization.Serialization.WithTransport(Serialization.System, () => serializer.FromBinary(binary, type));
             }
             else
             {

src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs

@@ -14,6 +14,7 @@
 using Akka.Event;
 using Akka.Persistence.Snapshot;
 using Akka.Util;
+using Akka.Util.Internal;
 
 namespace Akka.Persistence.Sql.Common.Snapshot
 {
@@ -39,12 +40,15 @@ private Initialized() { }
 
         private readonly SnapshotStoreSettings _settings;
 
+        private readonly ExtendedActorSystem _actorSystem;
+
         /// <summary>
         /// Initializes a new instance of the <see cref="SqlSnapshotStore"/> class.
         /// </summary>
         /// <param name="config">The configuration used to configure the snapshot store.</param>
         protected SqlSnapshotStore(Config config)
         {
+            _actorSystem = Context.System.AsInstanceOf<ExtendedActorSystem>();
             _settings = new SnapshotStoreSettings(config);
             _pendingRequestsCancellation = new CancellationTokenSource();
         }
@@ -224,7 +228,9 @@ private SnapshotEntry ToSnapshotEntry(SnapshotMetadata metadata, object snapshot
             var snapshotType = snapshot.GetType();
             var serializer = Context.System.Serialization.FindSerializerForType(snapshotType, _settings.DefaultSerializer);
 
-            var binary = serializer.ToBinary(snapshot);
+            var binary  = Akka.Serialization.Serialization.WithTransport(_actorSystem,
+                () => serializer.ToBinary(snapshot));
+
 
             return new SnapshotEntry(
                 persistenceId: metadata.PersistenceId,

src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Akka.Persistence.Sqlite.Tests.csproj

@@ -4,7 +4,7 @@
   <PropertyGroup>
     <AssemblyTitle>Akka.Persistence.Sqlite.Tests</AssemblyTitle>
     <TargetFrameworks>$(NetCoreTestVersion)</TargetFrameworks>
-    <RuntimeIdentifier Condition=" '$(TargetFramework)' == '$(NetFrameworkTestVersion)' And '$(OS)' == 'Windows_NT'">win7-x64</RuntimeIdentifier>
+
   </PropertyGroup>
 
   <ItemGroup>

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

@@ -42,5 +42,8 @@ class = ""Akka.Persistence.Sqlite.Journal.BatchingSqliteJournal, Akka.Persistenc
                     }
                 }");
         }
+
+        // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+        protected override bool SupportsSerialization => false;
     }
 }

src/contrib/persistence/Akka.Persistence.Sqlite.Tests/SqliteJournalSpec.cs

@@ -24,6 +24,9 @@ public SqliteJournalSpec(ITestOutputHelper output)
             Initialize();
         }
 
+        // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+        protected override bool SupportsSerialization => false;
+
         private static Config CreateSpecConfig(string connectionString)
         {
             return ConfigurationFactory.ParseString(@"

src/contrib/persistence/Akka.Persistence.Sqlite.Tests/SqliteSnapshotStoreSpec.cs

@@ -40,5 +40,7 @@ class = ""Akka.Persistence.Sqlite.Snapshot.SqliteSnapshotStore, Akka.Persistence
                     }
                 }");
         }
+
+        protected override bool SupportsSerialization => true;
     }
 }

src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt

@@ -6,6 +6,7 @@
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Cluster.Tools")]
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.MultiNodeTestRunner.Shared.Tests")]
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Persistence")]
+[assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Persistence.TCK")]
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Remote")]
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Remote.Tests")]
 [assembly: System.Runtime.CompilerServices.InternalsVisibleToAttribute("Akka.Remote.Tests.MultiNode")]
@@ -985,6 +986,8 @@ namespace Akka.Actor
         Akka.Actor.LocalActorRef Guardian { get; }
         Akka.Actor.IInternalActorRef RootGuardian { get; }
         Akka.Actor.ActorPath RootPath { get; }
+        [Akka.Annotations.InternalApiAttribute()]
+        Akka.Serialization.Information SerializationInformation { get; }
         Akka.Actor.Settings Settings { get; }
         Akka.Actor.LocalActorRef SystemGuardian { get; }
         Akka.Actor.IInternalActorRef TempContainer { get; }
@@ -1270,6 +1273,7 @@ namespace Akka.Actor
         public Akka.Event.ILoggingAdapter Log { get; }
         public Akka.Actor.IInternalActorRef RootGuardian { get; }
         public Akka.Actor.ActorPath RootPath { get; }
+        public Akka.Serialization.Information SerializationInformation { get; }
         public Akka.Actor.Settings Settings { get; }
         public Akka.Actor.LocalActorRef SystemGuardian { get; }
         public Akka.Actor.IInternalActorRef TempContainer { get; }
@@ -4541,6 +4545,16 @@ namespace Akka.Serialization
         public override object FromBinary(byte[] bytes, System.Type type) { }
         public override byte[] ToBinary(object obj) { }
     }
+    [Akka.Annotations.InternalApiAttribute()]
+    public sealed class Information : System.IEquatable<Akka.Serialization.Information>
+    {
+        public Information(Akka.Actor.Address address, Akka.Actor.ActorSystem system) { }
+        public Akka.Actor.Address Address { get; }
+        public Akka.Actor.ActorSystem System { get; }
+        public bool Equals(Akka.Serialization.Information other) { }
+        public override bool Equals(object obj) { }
+        public override int GetHashCode() { }
+    }
     public class NewtonSoftJsonSerializer : Akka.Serialization.Serializer
     {
         public NewtonSoftJsonSerializer(Akka.Actor.ExtendedActorSystem system) { }
@@ -4572,7 +4586,7 @@ namespace Akka.Serialization
     public class Serialization
     {
         public Serialization(Akka.Actor.ExtendedActorSystem system) { }
-        public Akka.Actor.ActorSystem System { get; }
+        public Akka.Actor.ExtendedActorSystem System { get; }
         public void AddSerializationMap(System.Type type, Akka.Serialization.Serializer serializer) { }
         [System.ObsoleteAttribute("No longer supported. Use the AddSerializer(name, serializer) overload instead.", true)]
         public void AddSerializer(Akka.Serialization.Serializer serializer) { }
@@ -4581,8 +4595,12 @@ namespace Akka.Serialization
         public object Deserialize(byte[] bytes, int serializerId, string manifest) { }
         public Akka.Serialization.Serializer FindSerializerFor(object obj, string defaultSerializerName = null) { }
         public Akka.Serialization.Serializer FindSerializerForType(System.Type objectType, string defaultSerializerName = null) { }
+        public static Akka.Serialization.Information GetCurrentTransportInformation() { }
+        public byte[] Serialize(object o) { }
         public static string SerializedActorPath(Akka.Actor.IActorRef actorRef) { }
-        public static T SerializeWithTransport<T>(Akka.Actor.ActorSystem system, Akka.Actor.Address address, System.Func<T> action) { }
+        [System.ObsoleteAttribute("Obsolete. Use the SerializeWithTransport<T>(ExtendedActorSystem) method instead.")]
+        public static T WithTransport<T>(Akka.Actor.ActorSystem system, Akka.Actor.Address address, System.Func<T> action) { }
+        public static T WithTransport<T>(Akka.Actor.ExtendedActorSystem system, System.Func<T> action) { }
     }
     public abstract class Serializer
     {

src/core/Akka.API.Tests/CoreAPISpec.ApproveRemote.approved.txt

@@ -188,6 +188,7 @@ namespace Akka.Remote
         public Akka.Actor.IActorRef RemoteWatcher { get; }
         public Akka.Actor.IInternalActorRef RootGuardian { get; }
         public Akka.Actor.ActorPath RootPath { get; }
+        public Akka.Serialization.Information SerializationInformation { get; }
         public Akka.Actor.Settings Settings { get; }
         public Akka.Actor.LocalActorRef SystemGuardian { get; }
         public Akka.Actor.IInternalActorRef TempContainer { get; }

src/core/Akka.Persistence.TCK.Tests/LocalSnapshotStoreSpec.cs

@@ -30,6 +30,8 @@ public LocalSnapshotStoreSpec(ITestOutputHelper output)
             Initialize();
         }
 
+        protected override bool SupportsSerialization => true;
+
         protected override void Dispose(bool disposing)
         {
             base.Dispose(disposing);

src/core/Akka.Persistence.TCK.Tests/MemoryJournalSpec.cs

@@ -20,5 +20,7 @@ public MemoryJournalSpec(ITestOutputHelper output)
         }
 
         protected override bool SupportsRejectingNonSerializableObjects { get { return false; } }
+
+        protected override bool SupportsSerialization => true;
     }
 }

src/core/Akka.Persistence.TCK.Tests/MemorySnapshotStoreSpec.cs

@@ -32,6 +32,8 @@ public MemorySnapshotStoreSpec(ITestOutputHelper output)
             Initialize();
         }
 
+        protected override bool SupportsSerialization => true;
+
         [Fact]
         public void MemorySnapshotStore_is_threadsafe()
         {