Improved Selector Caching

src/Fluss.HotChocolate/Fluss.HotChocolate.csproj

@@ -11,7 +11,6 @@
         <RepositoryUrl>https://github.com/atmina/fluss</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
         <PackageLicenseExpression>MIT</PackageLicenseExpression>
-        <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     </PropertyGroup>
 
     <ItemGroup>

src/Fluss.Regen/FileBuilders/SelectorFileBuilder.cs

@@ -39,11 +39,14 @@ public void WriteClassHeader()
 
     public void WriteEndNamespace()
     {
-        _writer.WriteIndentedLine("private record CacheEntryValue(object Value, global::System.Collections.Generic.IReadOnlyList<global::Fluss.UnitOfWorkRecordingProxy.EventListenerTypeWithKeyAndVersion>? EventListeners);");
+        _writer.WriteIndentedLine("private record CacheEntryValue(object Value, global::System.Collections.Generic.IReadOnlyList<global::Fluss.UnitOfWorkRecordingProxy.EventListenerTypeWithKeyAndVersion>? EventListeners, long CreatedAtVersion);");
         _writer.WriteLine();
         _writer.WriteIndented("private static async global::System.Threading.Tasks.ValueTask<bool> MatchesEventListenerState(global::Fluss.IUnitOfWork unitOfWork, CacheEntryValue value) ");
         using (_writer.WriteBraces())
         {
+            _writer.WriteIndentedLine("var version = await unitOfWork.ConsistentVersion();");
+            _writer.WriteIndentedLine("if (value.CreatedAtVersion == version) return true;");
+
             _writer.WriteIndented("foreach (var eventListenerData in value.EventListeners ?? global::System.Array.Empty<global::Fluss.UnitOfWorkRecordingProxy.EventListenerTypeWithKeyAndVersion>()) ");
             using (_writer.WriteBraces())
             {
@@ -152,7 +155,7 @@ public void WriteMethodCacheMiss(string returnType)
 
         using (_writer.WriteBraces())
         {
-            _writer.WriteIndentedLine("entry.Value = new CacheEntryValue(result, recordingUnitOfWork.GetRecordedListeners());");
+            _writer.WriteIndentedLine("entry.Value = new CacheEntryValue(result, recordingUnitOfWork.GetRecordedListeners(), await unitOfWork.ConsistentVersion());");
             _writer.WriteIndentedLine("entry.Size = 1;");
         }
 

src/Fluss.UnitTest/.gitignore

@@ -0,0 +1,2 @@
+TestResults/
+

src/Fluss/UnitOfWork/IUnitOfWork.cs

@@ -34,4 +34,5 @@ ValueTask<TReadModel> UnsafeGetReadModelWithoutAuthorization<TReadModel>(long? a
         where TKey : notnull where TReadModel : EventListener, IReadModel, IEventListenerWithKey<TKey>, new();
 
     IUnitOfWork WithPrefilledVersion(long? version);
+    IUnitOfWork CopyWithVersion(long version);
 }

src/Fluss/UnitOfWork/UnitOfWork.cs

@@ -90,6 +90,14 @@ public IUnitOfWork WithPrefilledVersion(long? version)
         return this;
     }
 
+    public IUnitOfWork CopyWithVersion(long version)
+    {
+        EnsureInstantiated();
+
+        var unitOfWork = Create(_eventRepository!, _eventListenerFactory!, _policies, _userIdProvider!, _validator!);
+        return unitOfWork.WithPrefilledVersion(version);
+    }
+
     private Guid CurrentUserId()
     {
         EnsureInstantiated();

src/Fluss/UnitOfWork/UnitOfWorkRecordingProxy.cs

@@ -1,10 +1,11 @@
-using System.Collections.Concurrent;
-using System.Diagnostics.CodeAnalysis;
+using System.Diagnostics.CodeAnalysis;
 using Fluss.Events;
 using Fluss.ReadModel;
 
 namespace Fluss;
 
+public class AtNotAllowedInSelectorException : Exception;
+
 public class UnitOfWorkRecordingProxy(IUnitOfWork impl) : IUnitOfWork
 {
     public ValueTask<long> ConsistentVersion()
@@ -18,42 +19,54 @@ public ValueTask<long> ConsistentVersion()
 
     public ValueTask<IReadModel> GetReadModel([DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicParameterlessConstructor)] Type tReadModel, object? key, long? at = null)
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return impl.GetReadModel(tReadModel, key, at);
     }
 
     public ValueTask<TReadModel> GetReadModel<TReadModel>(long? at = null) where TReadModel : EventListener, IRootEventListener, IReadModel, new()
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.GetReadModel<TReadModel>(at));
     }
 
     public ValueTask<TReadModel> GetReadModel<TReadModel, TKey>(TKey key, long? at = null) where TReadModel : EventListener, IEventListenerWithKey<TKey>, IReadModel, new()
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.GetReadModel<TReadModel, TKey>(key, at));
     }
 
     public ValueTask<TReadModel> UnsafeGetReadModelWithoutAuthorization<TReadModel>(long? at = null) where TReadModel : EventListener, IRootEventListener, IReadModel, new()
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.UnsafeGetReadModelWithoutAuthorization<TReadModel>(at));
     }
 
     public ValueTask<TReadModel> UnsafeGetReadModelWithoutAuthorization<TReadModel, TKey>(TKey key, long? at = null) where TReadModel : EventListener, IEventListenerWithKey<TKey>, IReadModel, new()
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.UnsafeGetReadModelWithoutAuthorization<TReadModel, TKey>(key, at));
     }
 
     public ValueTask<IReadOnlyList<TReadModel>> GetMultipleReadModels<TReadModel, TKey>(IEnumerable<TKey> keys, long? at = null) where TReadModel : EventListener, IReadModel, IEventListenerWithKey<TKey>, new() where TKey : notnull
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.GetMultipleReadModels<TReadModel, TKey>(keys, at));
     }
 
     public ValueTask<IReadOnlyList<TReadModel>> UnsafeGetMultipleReadModelsWithoutAuthorization<TReadModel, TKey>(IEnumerable<TKey> keys, long? at = null) where TReadModel : EventListener, IReadModel, IEventListenerWithKey<TKey>, new() where TKey : notnull
     {
+        if (at.HasValue) throw new AtNotAllowedInSelectorException();
         return Record(impl.UnsafeGetMultipleReadModelsWithoutAuthorization<TReadModel, TKey>(keys, at));
     }
 
     public IUnitOfWork WithPrefilledVersion(long? version)
     {
-        return impl.WithPrefilledVersion(version);
+        throw new AtNotAllowedInSelectorException();
+    }
+
+    public IUnitOfWork CopyWithVersion(long version)
+    {
+        throw new AtNotAllowedInSelectorException();
     }
 
     private async ValueTask<TReadModel> Record<TReadModel>(ValueTask<TReadModel> readModel) where TReadModel : EventListener