Feature/durable

.changes/.version

@@ -0,0 +1 @@
+1.0.1

CHANGELOG.md

@@ -0,0 +1,5 @@
+## 1.0.1
+
+- Updates to .NET 10
+- Updated to Durable State Implementation
+- Reworked Outbox

src/Strata.Journaling.Tests/JournalingTests/GrainModel/IDelayedAccountGrain.cs

@@ -0,0 +1,16 @@
+using Strata.Journaling.Tests.JournalingTests.Events;
+
+namespace Strata.Journaling.Tests.JournalingTests.GrainModel;
+
+public interface IDelayedAccountGrain : IGrainWithStringKey
+{
+    Task Deposit(double amount);
+
+    Task Withdraw(double amount);
+
+    Task<double> GetBalance();
+
+    Task<bool> GetIsProcessingOutbox();
+
+    Task<BaseAccountEvent[]> GetEvents();
+}

src/Strata.Journaling.Tests/JournalingTests/Grains/AccountGrain.cs

@@ -1,4 +1,5 @@
-using Strata.Journaling.Tests.JournalingTests.Events;
+using Microsoft.Extensions.Logging;
+using Strata.Journaling.Tests.JournalingTests.Events;
 using Strata.Journaling.Tests.JournalingTests.GrainModel;
 using Strata.Journaling.Tests.JournalingTests.Model;
 using Strata.Journaling.Tests.JournalingTests.Projections;
@@ -10,11 +11,18 @@ internal sealed class AccountGrain :
     JournaledGrain<AccountAggregate, BaseAccountEvent>,
     IAccountGrain
 {
+    private readonly ILogger<IAccountGrain> _logger;
+
+    public AccountGrain(ILogger<IAccountGrain> logger)
+    {
+        _logger = logger;
+    }
+
     protected override void OnRegisterRecipients()
     {
         RegisterRecipient(
             nameof(AccountProjection),
-            new AccountProjection(this.GrainFactory)
+            new AccountProjection(this.GrainFactory, _logger)
         );
     }
 
@@ -27,13 +35,13 @@ public ValueTask Deactivate()
     public override async Task OnActivateAsync(CancellationToken cancellationToken)
     {
         await base.OnActivateAsync(cancellationToken);
-        Console.WriteLine("[{0}] OnActivateAsync", this.GetPrimaryKeyString());
+        _logger.LogInformation("[{0}] OnActivateAsync", this.GetPrimaryKeyString());
     }
 
     public override async Task OnDeactivateAsync(DeactivationReason reason, CancellationToken cancellationToken)
     {
         await base.OnDeactivateAsync(reason, cancellationToken);
-        Console.WriteLine("[{0}] OnDeactivateAsync", this.GetPrimaryKeyString());
+        _logger.LogInformation("[{0}] OnDeactivateAsync", this.GetPrimaryKeyString());
     }
 
     public Task<BaseAccountEvent[]> GetEvents() => Task.FromResult(Log.Select(e => e.Event).ToArray());

src/Strata.Journaling.Tests/JournalingTests/Grains/AccountViewModelGrain.cs

@@ -1,4 +1,5 @@
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Strata.Journaling.Tests.JournalingTests.GrainModel;
 using Strata.Journaling.Tests.JournalingTests.Model;
 
@@ -8,11 +9,14 @@ internal sealed class AccountViewModelGrain : Grain, IAccountViewModelGrain
 {
     private readonly IPersistentState<AccountViewModel> _state;
 
+    private readonly ILogger<IAccountViewModelGrain> _logger;
+
     public AccountViewModelGrain(
-        [FromKeyedServices("state")] IPersistentState<AccountViewModel> state
-    )
+        [FromKeyedServices("state")] IPersistentState<AccountViewModel> state,
+        ILogger<IAccountViewModelGrain> logger)
     {
         _state = state;
+        _logger = logger;
     }
 
     public override async Task OnActivateAsync(CancellationToken cancellationToken)
@@ -31,7 +35,7 @@ public override async Task OnActivateAsync(CancellationToken cancellationToken)
 
     public async Task UpdateBalance(double newBalance)
     {
-        Console.WriteLine("Receiving balance update for account {0} to {1}", this.GetPrimaryKeyString(), newBalance);
+        _logger.LogInformation("Receiving balance update for account {0} to {1}", this.GetPrimaryKeyString(), newBalance);
         _state.State.Balance = newBalance;
         await _state.WriteStateAsync();
     }

src/Strata.Journaling.Tests/JournalingTests/Grains/DelayedAccountGrain.cs

@@ -0,0 +1,57 @@
+using Microsoft.Extensions.Logging;
+using Strata.Journaling.Tests.JournalingTests.Events;
+using Strata.Journaling.Tests.JournalingTests.GrainModel;
+using Strata.Journaling.Tests.JournalingTests.Model;
+using Strata.Journaling.Tests.JournalingTests.Projections;
+
+namespace Strata.Journaling.Tests.JournalingTests.Grains;
+
+[GrainType("delayed-account")]
+internal sealed class DelayedAccountGrain :
+    JournaledGrain<AccountAggregate, BaseAccountEvent>,
+    IDelayedAccountGrain
+{
+    private readonly ILogger<IDelayedAccountGrain> _logger;
+
+    public DelayedAccountGrain(ILogger<IDelayedAccountGrain> logger)
+    {
+        _logger = logger;
+    }
+
+    protected override void OnRegisterRecipients()
+    {
+        RegisterRecipient(
+            nameof(DelayedAccountProjection),
+            new DelayedAccountProjection(this.GrainFactory, _logger)
+        );
+    }
+
+    public Task<BaseAccountEvent[]> GetEvents() => Task.FromResult(Log.Select(e => e.Event).ToArray());
+
+    public Task<double> GetBalance() => Task.FromResult(ConfirmedState.Balance);
+
+    public Task<bool> GetIsProcessingOutbox() => Task.FromResult(IsProcessingOutbox);
+
+    public async Task Deposit(double amount)
+    {
+        if (amount <= 0) throw new ArgumentOutOfRangeException(nameof(amount), "Deposit amount must be positive.");
+        var newBalance = ConfirmedState.Balance + amount;
+        var @event = new BalanceAdjustedEvent(this.GetPrimaryKeyString()) { Balance = newBalance };
+
+        _logger.LogInformation("Depositing {0} to account {1}. New balance will be {2}", amount, this.GetPrimaryKeyString(), newBalance);
+
+        await RaiseEvent(@event);
+    }
+
+    public async Task Withdraw(double amount)
+    {
+        if (amount <= 0) throw new ArgumentOutOfRangeException(nameof(amount), "Withdrawal amount must be positive.");
+        if (amount > ConfirmedState.Balance) throw new InvalidOperationException("Insufficient funds for withdrawal.");
+        var newBalance = ConfirmedState.Balance - amount;
+        var @event = new BalanceAdjustedEvent(this.GetPrimaryKeyString()) { Balance = newBalance };
+
+        _logger.LogInformation("Withdrawing {0} from account {1}. New balance will be {2}", amount, this.GetPrimaryKeyString(), newBalance);
+
+        await RaiseEvent(@event);
+    }
+}

src/Strata.Journaling.Tests/JournalingTests/JournaledGrainTests.cs

@@ -87,4 +87,36 @@ public async Task JournaledGrain_MultipleGrains()
         Assert.Equal(100, balance1);
         Assert.Equal(200, balance2);
     }
+
+    [Fact]
+    public async Task JournaledGrain_DelayedProjection_OutboxClearsAfterStackedOperations()
+    {
+        var accountId = "delayed_projection_outbox_stacking";
+        var grain = Client.GetGrain<IDelayedAccountGrain>(accountId);
+        var projectionGrain = Client.GetGrain<IAccountViewModelGrain>(accountId);
+
+        await grain.Deposit(200);
+        await grain.Withdraw(25);
+        await grain.Deposit(40);
+        await grain.Withdraw(15);
+        await grain.Deposit(10);
+
+        var timeoutAt = DateTime.UtcNow.AddSeconds(15);
+        while (DateTime.UtcNow < timeoutAt)
+        {
+            if (!await grain.GetIsProcessingOutbox())
+            {
+                break;
+            }
+
+            await Task.Delay(50);
+        }
+
+        Assert.False(await grain.GetIsProcessingOutbox());
+        Assert.Equal(210, await grain.GetBalance());
+        Assert.Equal(210, await projectionGrain.GetBalance());
+
+        var log = await grain.GetEvents();
+        Assert.Equal(5, log.Length);
+    }
 }

src/Strata.Journaling.Tests/JournalingTests/JournalingTestFixture.cs

@@ -1,5 +1,6 @@
 using System.Collections.Concurrent;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Orleans.Configuration;
 using Orleans.Journaling;
 using Orleans.TestingHost;
@@ -20,8 +21,20 @@ public JournalingTestFixture()
         var builder = new InProcessTestClusterBuilder();
         var storageProvider = new VolatileStateMachineStorageProvider();
 
+        static void ConfigureConsoleLogging(ILoggingBuilder logging)
+        {
+            logging.AddSimpleConsole(options =>
+            {
+                options.SingleLine = true;
+                options.TimestampFormat = "HH:mm:ss ";
+            });
+            logging.SetMinimumLevel(LogLevel.Information);
+        }
+
         builder.ConfigureSilo((options, siloBuilder) =>
         {
+            siloBuilder.UseInMemoryReminderService();
+            siloBuilder.ConfigureLogging(ConfigureConsoleLogging);
             siloBuilder.AddStateMachineStorage();
             siloBuilder.Services.AddSingleton<IStateMachineStorageProvider>(storageProvider);
 
@@ -31,6 +44,7 @@ public JournalingTestFixture()
                 options.CollectionAge = TimeSpan.FromSeconds(61);
             });
         });
+
         ConfigureTestCluster(builder);
         Cluster = builder.Build();
     }

src/Strata.Journaling.Tests/JournalingTests/Projections/AccountProjection.cs

@@ -1,15 +1,18 @@
-using Strata.Journaling.Tests.JournalingTests.Events;
+using Microsoft.Extensions.Logging;
+using Strata.Journaling.Tests.JournalingTests.Events;
 using Strata.Journaling.Tests.JournalingTests.GrainModel;
 
 namespace Strata.Journaling.Tests.JournalingTests.Projections;
 
 public sealed class AccountProjection : IOutboxRecipient<BaseAccountEvent>
 {
     private readonly IGrainFactory _grainFactory;
+    private readonly ILogger _logger;
 
-    public AccountProjection(IGrainFactory grainFactory)
+    public AccountProjection(IGrainFactory grainFactory, ILogger logger)
     {
         _grainFactory = grainFactory;
+        _logger = logger;
     }
 
     public async Task Handle(int version, BaseAccountEvent @event)
@@ -18,7 +21,7 @@ public async Task Handle(int version, BaseAccountEvent @event)
         {
             var accountId = balanceEvent.Id;
 
-            Console.WriteLine("Updating balance for account {0} to {1}", accountId, balanceEvent.Balance);
+            _logger.LogInformation("Updating balance for account {0} to {1}", accountId, balanceEvent.Balance);
 
             var viewModelGrain = _grainFactory.GetGrain<IAccountViewModelGrain>(accountId);
             await viewModelGrain.UpdateBalance(balanceEvent.Balance);

src/Strata.Journaling.Tests/JournalingTests/Projections/DelayedAccountProjection.cs

@@ -0,0 +1,34 @@
+using Microsoft.Extensions.Logging;
+using Strata.Journaling.Tests.JournalingTests.Events;
+using Strata.Journaling.Tests.JournalingTests.GrainModel;
+
+namespace Strata.Journaling.Tests.JournalingTests.Projections;
+
+public sealed class DelayedAccountProjection : IOutboxRecipient<BaseAccountEvent>
+{
+    private static readonly TimeSpan ProjectionDelay = TimeSpan.FromMilliseconds(250);
+
+    private readonly IGrainFactory _grainFactory;
+    private readonly ILogger _logger;
+
+    public DelayedAccountProjection(IGrainFactory grainFactory, ILogger logger)
+    {
+        _grainFactory = grainFactory;
+        _logger = logger;
+    }
+
+    public async Task Handle(int version, BaseAccountEvent @event)
+    {
+        if (@event is BalanceAdjustedEvent balanceEvent)
+        {
+            var accountId = balanceEvent.Id;
+
+            await Task.Delay(ProjectionDelay);
+
+            _logger.LogInformation("[Delayed] Updating balance for account {0} to {1}", accountId, balanceEvent.Balance);
+
+            var viewModelGrain = _grainFactory.GetGrain<IAccountViewModelGrain>(accountId);
+            await viewModelGrain.UpdateBalance(balanceEvent.Balance);
+        }
+    }
+}