Support sync all observers on activate.

Author: u-less

src/Ray.Core.Abstractions/Observer/IVersion.cs

@@ -6,5 +6,6 @@ public interface IVersion
     {
         Task<long> GetVersion();
         Task<long> GetAndSaveVersion(long compareVersion);
+        Task<bool> SyncFromObservable(long compareVersion);
     }
 }

src/Ray.Core/Configuration/CoreOptions.cs

@@ -31,5 +31,9 @@ public class CoreOptions
         /// 事务超时时间,单位为ms(默认为30s)
         /// </summary>
         public int TransactionTimeout { get; set; } = 30 * 1000;
+        /// <summary>
+        /// 在激活的时候同步所有观察者
+        /// </summary>
+        public bool SyncAllObserversOnActivate { get; set; } = false;
     }
 }

src/Ray.Core/Core/Abstractions/IObserverUnit.cs

@@ -6,6 +6,13 @@ namespace Ray.Core.Abstractions
 {
     public interface IObserverUnit<PrimaryKey> : IGrainID
     {
+        /// <summary>
+        /// 同步所有观察者
+        /// </summary>
+        /// <param name="primaryKey">GrainId</param>
+        /// <param name="srcVersion">Observable的Version</param>
+        /// <returns></returns>
+        Task<bool[]> SyncAllObservers(PrimaryKey primaryKey, long srcVersion);
         /// <summary>
         /// 获取所有监听者分组
         /// </summary>

src/Ray.Core/Core/Grains/ObserverGrain.cs

@@ -314,12 +314,7 @@ public override async Task OnActivateAsync()
                 await ReadSnapshotAsync();
                 if (FullyActive)
                 {
-                    while (true)
-                    {
-                        var eventList = await EventStorage.GetList(GrainId, Snapshot.StartTimestamp, Snapshot.Version + 1, Snapshot.Version + ConfigOptions.NumberOfEventsPerRead);
-                        await UnsafeTell(eventList);
-                        if (eventList.Count < ConfigOptions.NumberOfEventsPerRead) break;
-                    };
+                    await RecoveryFromStorage();
                 }
                 if (Logger.IsEnabled(LogLevel.Trace))
                     Logger.LogTrace("Activation completed: {0}->{1}", GrainType.FullName, Serializer.Serialize(Snapshot));
@@ -330,6 +325,19 @@ public override async Task OnActivateAsync()
                 throw;
             }
         }
+        /// <summary>
+        /// 从库里恢复
+        /// </summary>
+        /// <returns></returns>
+        private async Task RecoveryFromStorage()
+        {
+            while (true)
+            {
+                var eventList = await EventStorage.GetList(GrainId, Snapshot.StartTimestamp, Snapshot.Version + 1, Snapshot.Version + ConfigOptions.NumberOfEventsPerRead);
+                await UnsafeTell(eventList);
+                if (eventList.Count < ConfigOptions.NumberOfEventsPerRead) break;
+            };
+        }
         public override Task OnDeactivateAsync()
         {
             var needSaveSnap = Snapshot.Version - SnapshotEventVersion >= 1;
@@ -515,6 +523,14 @@ public async Task<long> GetAndSaveVersion(long compareVersion)
             }
             return Snapshot.Version;
         }
+        public async Task<bool> SyncFromObservable(long compareVersion)
+        {
+            if (Snapshot.Version < compareVersion)
+            {
+                await RecoveryFromStorage();
+            }
+            return Snapshot.Version == compareVersion;
+        }
         protected async ValueTask Tell(FullyEvent<PrimaryKey> fullyEvent)
         {
             try

src/Ray.Core/Core/Grains/RayGrain.cs

@@ -194,7 +194,15 @@ public override async Task OnActivateAsync()
                         };
                     }
                 }
-                var onActivatedTask = OnBaseActivated();
+                if (CoreOptions.SyncAllObserversOnActivate)
+                {
+                    var syncResults = await ObserverUnit.SyncAllObservers(GrainId, Snapshot.Base.Version);
+                    if (syncResults.Any(r => !r))
+                    {
+                        throw new SyncAllObserversException(GrainId.ToString(), GrainType);
+                    }
+                }
+                var onActivatedTask = OnActivationCompleted();
                 if (!onActivatedTask.IsCompletedSuccessfully)
                     await onActivatedTask;
                 if (Logger.IsEnabled(LogLevel.Trace))
@@ -207,7 +215,7 @@ public override async Task OnActivateAsync()
             }
         }
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        protected virtual ValueTask OnBaseActivated() => Consts.ValueTaskDone;
+        protected virtual ValueTask OnActivationCompleted() => Consts.ValueTaskDone;
         protected virtual async Task RecoverySnapshot()
         {
             try

src/Ray.Core/Core/Grains/ShadowGrain.cs

@@ -138,14 +138,7 @@ public override async Task OnActivateAsync()
                 await ReadSnapshotAsync();
                 if (FullyActive)
                 {
-                    while (true)
-                    {
-                        var eventList = await EventStorage.GetList(GrainId, Snapshot.Base.StartTimestamp, Snapshot.Base.Version + 1, Snapshot.Base.Version + NumberOfEventsPerRead);
-                        var task = Tell(eventList);
-                        if (!task.IsCompletedSuccessfully)
-                            await task;
-                        if (eventList.Count < NumberOfEventsPerRead) break;
-                    };
+                    await RecoveryFromStorage();
                 }
                 if (Logger.IsEnabled(LogLevel.Trace))
                     Logger.LogTrace("Activation completed: {0}->{1}", GrainType.FullName, Serializer.Serialize(Snapshot));
@@ -156,6 +149,21 @@ public override async Task OnActivateAsync()
                 throw;
             }
         }
+        /// <summary>
+        /// 从库里恢复
+        /// </summary>
+        /// <returns></returns>
+        private async Task RecoveryFromStorage()
+        {
+            while (true)
+            {
+                var eventList = await EventStorage.GetList(GrainId, Snapshot.Base.StartTimestamp, Snapshot.Base.Version + 1, Snapshot.Base.Version + NumberOfEventsPerRead);
+                var task = Tell(eventList);
+                if (!task.IsCompletedSuccessfully)
+                    await task;
+                if (eventList.Count < NumberOfEventsPerRead) break;
+            };
+        }
         protected virtual async ValueTask Tell(IEnumerable<FullyEvent<PrimaryKey>> eventList)
         {
             foreach (var @event in eventList)
@@ -298,6 +306,14 @@ public Task<long> GetAndSaveVersion(long compareVersion)
         {
             return Task.FromResult(Snapshot.Base.Version);
         }
+        public async Task<bool> SyncFromObservable(long compareVersion)
+        {
+            if (Snapshot.Base.Version < compareVersion)
+            {
+                await RecoveryFromStorage();
+            }
+            return Snapshot.Base.Version == compareVersion;
+        }
         protected async ValueTask Tell(FullyEvent<PrimaryKey> @event)
         {
             if (@event.Base.Version == Snapshot.Base.Version + 1)

src/Ray.Core/Core/Observer/ObserverUnit.cs

@@ -27,6 +27,7 @@ public class ObserverUnit<PrimaryKey> : IObserverUnit<PrimaryKey>
         readonly List<Func<List<byte[]>, Task>> batchEventHandlers = new List<Func<List<byte[]>, Task>>();
         readonly List<Func<PrimaryKey, long, Task<long>>> observerVersionHandlers = new List<Func<PrimaryKey, long, Task<long>>>();
         readonly List<Func<PrimaryKey, Task>> observerResetHandlers = new List<Func<PrimaryKey, Task>>();
+        readonly List<Func<PrimaryKey, long, Task<bool>>> observerSyncHandlers = new List<Func<PrimaryKey, long, Task<bool>>>();
         protected ILogger Logger { get; private set; }
         public Type GrainType { get; }
 
@@ -47,7 +48,10 @@ public Task<long[]> GetAndSaveVersion(PrimaryKey primaryKey, long srcVersion)
         {
             return Task.WhenAll(observerVersionHandlers.Select(func => func(primaryKey, srcVersion)));
         }
-
+        public Task<bool[]> SyncAllObservers(PrimaryKey primaryKey, long srcVersion)
+        {
+            return Task.WhenAll(observerSyncHandlers.Select(func => func(primaryKey, srcVersion)));
+        }
         public Task Reset(PrimaryKey primaryKey)
         {
             return Task.WhenAll(observerResetHandlers.Select(func => func(primaryKey)));
@@ -149,6 +153,7 @@ public void Observer(string group, Type observerType)
             eventHandlers.Add(EventHandler);
             batchEventHandlers.Add(BatchEventHandler);
             observerVersionHandlers.Add((actorId, version) => GetObserver(observerType, actorId).GetAndSaveVersion(version));
+            observerSyncHandlers.Add((actorId, version) => GetObserver(observerType, actorId).SyncFromObservable(version));
             observerResetHandlers.Add((actorId) => GetObserver(observerType, actorId).Reset());
             //内部函数
             Task EventHandler(byte[] bytes)

src/Ray.Core/Exceptions/SyncAllObserversException.cs

@@ -0,0 +1,12 @@
+using System;
+
+namespace Ray.Core.Exceptions
+{
+    public class SyncAllObserversException : Exception
+    {
+        public SyncAllObserversException(string id, Type grainType) :
+            base($"Sync all observers failed of Grain type {grainType.FullName} and Id {id}")
+        {
+        }
+    }
+}