First class migration support for switching a projection from Inline to Async. Closes GH-3564

Author: jeremydmiller

docs/configuration/cli.md

@@ -33,7 +33,7 @@ And finally, use Oakton as the command line parser and executor by replacing `Ap
 // as the last line of your Program.cs file
 return await app.RunOaktonCommands(args);
 ```
-<sup><a href='https://github.com/JasperFx/marten/blob/master/src/samples/MinimalAPI/Program.cs#L51-L57' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_using_webapplication_2' title='Start of snippet'>anchor</a></sup>
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/samples/MinimalAPI/Program.cs#L55-L61' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_using_webapplication_2' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
 Once the _Marten.CommandLine_ Nuget is installed and Oakton is handling your command line parsing, you should be able to see the Marten commands by typing `dotnet run -- help` in the command line terminal of your choice at the root of your project:

docs/events/projections/async-daemon.md

@@ -528,3 +528,40 @@ var latest = await session.QueryForNonStaleData<Trip>(5.Seconds())
 
 Do note that this can time out if the projection just can't catch up to the latest event sequence in time. You may need to
 be both cautious with using this in general, and also cautious especially with the timeout setting. 
+
+## Migrating a Projection from Inline to Async <Badge type="tip" text="7.35" />
+
+::: warning
+This will only work correctly *if* you have system downtime before migrating the new version of the code with this option
+enabled. This feature cannot support a "blue/green" deployment model. Marten needs to system to be at rest before it starts
+up the projection asynchronously or there's a chance you may "skip" events in the projection.
+:::
+
+During the course of a system's lifetime, you may find that you want to change a projection that's currently running
+with a lifecycle of `Inline` to running asynchronously instead. If you need to do this *and* there is no structural change
+to the projection that would require a projection rebuild, you can direct Marten to start that projection at the highest
+sequence number assigned by the system (not the high water mark, but the event sequence number which may be higher).
+
+To do so, use this option when registering the projection:
+
+<!-- snippet: sample_using_subscribe_as_inline_to_async -->
+<a id='snippet-sample_using_subscribe_as_inline_to_async'></a>
+```cs
+opts
+    .Projections
+    .Snapshot<SimpleAggregate>(SnapshotLifecycle.Async, o =>
+    {
+        // This option tells Marten to start the async projection at the highest
+        // event sequence assigned as the processing floor if there is no previous
+        // async daemon progress for this projection
+        o.SubscribeAsInlineToAsync();
+    });
+```
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/DaemonTests/converting_projection_from_inline_to_async.cs#L31-L43' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_using_subscribe_as_inline_to_async' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+Just to be clear, when Marten's async daemon starts a projection with this starting option:
+
+1. If there is no previously recorded progression, Marten will start processing this projection with the highest assigned event sequence
+   in the database as the floor and record that value as the current progress
+2. If there is a previously recorded progression, Marten will start processing this projection at the recorded sequence as normal

src/DaemonTests/AsyncOptionsTests.cs

@@ -239,6 +239,30 @@ public async Task subscribe_from_sequence_hit_with_prior_higher_than_threshold_a
             .ShouldBe(new Position(222L, true));
     }
 
+    [Fact]
+    public async Task transition_from_inline_to_async_no_initial_progress()
+    {
+        theDatabase.ProjectionProgressFor(theName, theToken).Returns(0);
+        theDatabase.FetchHighestEventSequenceNumber().Returns(1234L);
+        var options = new AsyncOptions();
+        options.SubscribeAsInlineToAsync();
+
+        (await options.DetermineStartingPositionAsync(1000L, theName, ShardExecutionMode.Continuous, theDatabase, theToken))
+            .ShouldBe(new Position(1234L, true));
+    }
+
+    [Fact]
+    public async Task transition_from_inline_to_async_but_there_is_initial_progress()
+    {
+        theDatabase.ProjectionProgressFor(theName, theToken).Returns(1000L);
+        theDatabase.FetchHighestEventSequenceNumber().Returns(2005L);
+        var options = new AsyncOptions();
+        options.SubscribeAsInlineToAsync();
+
+        (await options.DetermineStartingPositionAsync(2003L, theName, ShardExecutionMode.Continuous, theDatabase, theToken))
+            .ShouldBe(new Position(1000L, false));
+    }
+
 
 
 

src/DaemonTests/converting_projection_from_inline_to_async.cs

@@ -0,0 +1,155 @@
+using System;
+using System.Threading.Tasks;
+using DaemonTests.TestingSupport;
+using JasperFx.Core;
+using Marten.Events.Projections;
+using Marten.Metadata;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace DaemonTests;
+
+public class converting_projection_from_inline_to_async : OneOffConfigurationsContext
+{
+    [Fact]
+    public async Task start_as_inline_move_to_async_and_just_continue()
+    {
+        StoreOptions(opts =>
+        {
+            opts.Projections.Snapshot<SimpleAggregate>(SnapshotLifecycle.Inline);
+        });
+
+        var id1 = theSession.Events.StartStream<SimpleAggregate>(new AEvent(), new BEvent()).Id;
+        var id2 = theSession.Events.StartStream<SimpleAggregate>(new BEvent(), new CEvent()).Id;
+        var id3 = theSession.Events.StartStream<SimpleAggregate>(new CEvent(), new DEvent()).Id;
+        await theSession.SaveChangesAsync();
+
+        var store2 = SeparateStore(opts =>
+        {
+            #region sample_using_subscribe_as_inline_to_async
+
+            opts
+                .Projections
+                .Snapshot<SimpleAggregate>(SnapshotLifecycle.Async, o =>
+                {
+                    // This option tells Marten to start the async projection at the highest
+                    // event sequence assigned as the processing floor if there is no previous
+                    // async daemon progress for this projection
+                    o.SubscribeAsInlineToAsync();
+                });
+
+            #endregion
+        });
+
+        using var daemon = await store2.BuildProjectionDaemonAsync();
+        await daemon.StartAllAsync();
+
+        using var session = store2.LightweightSession();
+        session.Events.Append(id1, new EEvent(), new EEvent());
+        session.Events.Append(id2, new EEvent(), new EEvent());
+        session.Events.Append(id3, new EEvent(), new EEvent());
+        await session.SaveChangesAsync();
+
+        await daemon.WaitForNonStaleData(10.Seconds());
+
+        var aggregate1 = await session.LoadAsync<SimpleAggregate>(id1);
+        var aggregate2 = await session.LoadAsync<SimpleAggregate>(id2);
+        var aggregate3 = await session.LoadAsync<SimpleAggregate>(id3);
+
+        aggregate1.ShouldBe(new SimpleAggregate
+        {
+            Id = id1,
+            Version = 4,
+            ACount = 1,
+            BCount = 1,
+            ECount = 2
+        });
+
+        aggregate2.ShouldBe(new SimpleAggregate
+        {
+            Id = id2,
+            Version = 4,
+            BCount = 1,
+            CCount = 1,
+            ECount = 2
+        });
+
+    }
+}
+
+public class SimpleAggregate : IRevisioned
+{
+    // This will be the aggregate version
+    public int Version { get; set; }
+
+    public Guid Id { get; set; }
+
+    public int ACount { get; set; }
+    public int BCount { get; set; }
+    public int CCount { get; set; }
+    public int DCount { get; set; }
+    public int ECount { get; set; }
+
+    public void Apply(AEvent _)
+    {
+        ACount++;
+    }
+
+    public void Apply(BEvent _)
+    {
+        BCount++;
+    }
+
+    public void Apply(CEvent _)
+    {
+        CCount++;
+    }
+
+    public void Apply(DEvent _)
+    {
+        DCount++;
+    }
+
+    public void Apply(EEvent _)
+    {
+        ECount++;
+    }
+
+    public override string ToString()
+    {
+        return
+            $"{nameof(Version)}: {Version}, {nameof(Id)}: {Id}, {nameof(ACount)}: {ACount}, {nameof(BCount)}: {BCount}, {nameof(CCount)}: {CCount}, {nameof(DCount)}: {DCount}, {nameof(ECount)}: {ECount}";
+    }
+
+    protected bool Equals(SimpleAggregate other)
+    {
+        return Id.Equals(other.Id) && ACount == other.ACount && BCount == other.BCount && CCount == other.CCount && DCount == other.DCount && ECount == other.ECount;
+    }
+
+    public override bool Equals(object obj)
+    {
+        if (obj is null)
+        {
+            return false;
+        }
+
+        if (ReferenceEquals(this, obj))
+        {
+            return true;
+        }
+
+        if (obj.GetType() != GetType())
+        {
+            return false;
+        }
+
+        return Equals((SimpleAggregate)obj);
+    }
+
+    public override int GetHashCode()
+    {
+        return HashCode.Combine(Id, ACount, BCount, CCount, DCount, ECount);
+    }
+}

src/Marten/Events/Daemon/AsyncOptions.cs

@@ -149,6 +149,23 @@ public AsyncOptions SubscribeFromSequence(long sequenceFloor, string? databaseId
         _strategies.Add(new FromSequence(databaseIdentifier, sequenceFloor));
         return this;
     }
+
+    /// <summary>
+    /// Use this option to prevent having to rebuild a projection when you are
+    /// simply changing the projection lifecycle from "Inline" to "Async" but are
+    /// making no other changes that would force a rebuild
+    ///
+    /// Direct that this projection had previously been running with an "Inline"
+    /// lifecycle now run as "Async". This will cause Marten to first check if there
+    /// is any previous async progress, and if not, start the projection from the highest
+    /// event sequence for the system.
+    /// </summary>
+    /// <returns></returns>
+    public AsyncOptions SubscribeAsInlineToAsync()
+    {
+        _strategies.Add(new InlineToAsync());
+        return this;
+    }
 }
 
 internal record Position(long Floor, bool ShouldUpdateProgressFirst);
@@ -162,6 +179,24 @@ Task<Position> DetermineStartingPositionAsync(long highWaterMark, ShardName name
         CancellationToken token);
 }
 
+internal class InlineToAsync(): IPositionStrategy
+{
+    public string? DatabaseName => null;
+
+    public async Task<Position> DetermineStartingPositionAsync(long highWaterMark, ShardName name, ShardExecutionMode mode,
+        IMartenDatabase database, CancellationToken token)
+    {
+        var current = await database.ProjectionProgressFor(name, token).ConfigureAwait(false);
+        if (current > 0)
+        {
+            return new Position(current, false);
+        }
+
+        var highest = await database.FetchHighestEventSequenceNumber(token).ConfigureAwait(false);
+        return new Position(highest, true);
+    }
+}
+
 internal class FromSequence(string? databaseName, long sequence): IPositionStrategy
 {
     public string? DatabaseName { get; } = databaseName;