Initial version of query for non stale aggregate data.

Author: jeremydmiller

src/EventSourcingTests/Aggregation/querying_with_non_stale_data.cs

@@ -5,6 +5,7 @@
 using EventSourcingTests.FetchForWriting;
 using EventSourcingTests.Projections;
 using JasperFx.Core;
+using Marten;
 using Marten.Events;
 using Marten.Events.Daemon;
 using Marten.Events.Projections;
@@ -95,4 +96,32 @@ await theStore.Storage.Database.WaitForNonStaleProjectionDataAsync(typeof(Simple
         }
 
     }
+
+    [Fact]
+    public async Task try_to_query_for_non_stale_data_by_aggregate_type()
+    {
+        var task = Task.Run(async () =>
+        {
+            using var session = theStore.LightweightSession();
+
+            for (int i = 0; i < 20; i++)
+            {
+                session.Events.StartStream(new AEvent(), new BEvent());
+                session.Events.StartStream(new CEvent(), new BEvent());
+                session.Events.StartStream(new DEvent(), new AEvent());
+                var streamId = session.Events.StartStream(new DEvent(), new CEvent());
+                await session.SaveChangesAsync();
+            }
+        });
+
+        using var daemon = await theStore.BuildProjectionDaemonAsync();
+        await daemon.StartAllAsync();
+
+        await Task.Delay(1.Seconds());
+
+        var items = await theSession.QueryForNonStaleData<SimpleAggregate>(30.Seconds()).ToListAsync();
+        items.Count.ShouldBeGreaterThan(0);
+
+        await task;
+    }
 }

src/Marten/Events/AsyncProjectionTestingExtensions.cs

@@ -140,9 +140,7 @@ public static async Task WaitForNonStaleProjectionDataAsync(this IMartenDatabase
     {
         // Number of active projection shards, plus the high water mark
         var shards = database.As<MartenDatabase>().Options.Projections.AsyncShardsPublishingType(aggregationType);
-        if (!shards.Any())
-            throw new InvalidOperationException(
-                $"Cannot find any registered async projection shards for aggregate type {aggregationType.FullNameInCode()}");
+        if (!shards.Any()) throw new InvalidOperationException($"Cannot find any registered async projection shards for aggregate type {aggregationType.FullNameInCode()}");
 
         var all = shards.Concat([ShardName.HighWaterMark]).ToArray();
         var tracking = new Dictionary<string, long>();

src/Marten/IQuerySession.cs

@@ -165,6 +165,17 @@ public interface IQuerySession: IDisposable, IAsyncDisposable
 
     #endregion
 
+    /// <summary>
+    /// If you are querying for data against a projected event aggregation that is updated asynchronously
+    /// through the async daemon, this method will ensure that you are querying against the latest events appended
+    /// to the system by waiting for the aggregate to catch up to the current "high water mark" of the event store
+    /// at the time this query is executed.
+    /// </summary>
+    /// <param name="timeout"></param>
+    /// <typeparam name="T"></typeparam>
+    /// <returns></returns>
+    IMartenQueryable<T> QueryForNonStaleData<T>(TimeSpan timeout);
+
     /// <summary>
     ///     Queries the document storage table for the document type T by supplied SQL. See
     ///     https://martendb.io/documents/querying/sql.html for more information on usage.
@@ -763,4 +774,6 @@ Task<DocumentMetadata> MetadataForAsync<T>(T entity,
     /// <typeparam name="T"></typeparam>
     /// <returns></returns>
     Task<T> QueryByPlanAsync<T>(IQueryPlan<T> plan, CancellationToken token = default);
+
+
 }

src/Marten/Internal/Sessions/QuerySession.Querying.cs

@@ -1,4 +1,5 @@
 #nullable enable
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -22,6 +23,14 @@ public IMartenQueryable<T> Query<T>()
         return new MartenLinqQueryable<T>(this);
     }
 
+    public IMartenQueryable<T> QueryForNonStaleData<T>(TimeSpan timeout)
+    {
+        var queryable = new MartenLinqQueryable<T>(this);
+        queryable.MartenProvider.Waiter = new WaitForAggregate(timeout);
+
+        return queryable;
+    }
+
     public IReadOnlyList<T> Query<T>(string sql, params object[] parameters)
     {
         assertNotDisposed();

src/Marten/Linq/MartenLinqQueryProvider.cs

@@ -7,6 +7,7 @@
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
+using Marten.Events;
 using Marten.Exceptions;
 using Marten.Internal.Sessions;
 using Marten.Linq.Parsing;
@@ -16,6 +17,8 @@
 
 namespace Marten.Linq;
 
+internal record WaitForAggregate(TimeSpan Timeout);
+
 internal class MartenLinqQueryProvider: IQueryProvider
 {
     private readonly QuerySession _session;
@@ -28,6 +31,8 @@ public MartenLinqQueryProvider(QuerySession session, Type type)
 
     public Type SourceType { get; }
 
+    internal WaitForAggregate? Waiter { get; set; }
+
     internal QueryStatistics Statistics { get; set; } = null!;
 
     public IQueryable CreateQuery(Expression expression)
@@ -70,6 +75,11 @@ internal async ValueTask EnsureStorageExistsAsync(LinqQueryParser parser,
         {
             await _session.Database.EnsureStorageExistsAsync(documentType, cancellationToken).ConfigureAwait(false);
         }
+
+        if (Waiter != null)
+        {
+            await _session.Database.WaitForNonStaleProjectionDataAsync(SourceType, Waiter.Timeout, cancellationToken).ConfigureAwait(false);
+        }
     }