diff --git a/.changes/.version b/.changes/.version new file mode 100644 index 0000000..7f20734 --- /dev/null +++ b/.changes/.version @@ -0,0 +1 @@ +1.0.1 \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..7dc5f74 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +## 1.0.1 + +- Updates to .NET 10 +- Updated to Durable State Implementation +- Reworked Outbox diff --git a/src/Strata.Journaling.Tests/JournalingTests/GrainModel/IDelayedAccountGrain.cs b/src/Strata.Journaling.Tests/JournalingTests/GrainModel/IDelayedAccountGrain.cs new file mode 100644 index 0000000..1637ca1 --- /dev/null +++ b/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 GetBalance(); + + Task GetIsProcessingOutbox(); + + Task GetEvents(); +} diff --git a/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountGrain.cs b/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountGrain.cs index 55f8247..45dbfb6 100644 --- a/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountGrain.cs +++ b/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, IAccountGrain { + private readonly ILogger _logger; + + public AccountGrain(ILogger 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 GetEvents() => Task.FromResult(Log.Select(e => e.Event).ToArray()); diff --git a/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountViewModelGrain.cs b/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountViewModelGrain.cs index 3f10ec8..184641c 100644 --- a/src/Strata.Journaling.Tests/JournalingTests/Grains/AccountViewModelGrain.cs +++ b/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 _state; + private readonly ILogger _logger; + public AccountViewModelGrain( - [FromKeyedServices("state")] IPersistentState state - ) + [FromKeyedServices("state")] IPersistentState state, + ILogger 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(); } diff --git a/src/Strata.Journaling.Tests/JournalingTests/Grains/DelayedAccountGrain.cs b/src/Strata.Journaling.Tests/JournalingTests/Grains/DelayedAccountGrain.cs new file mode 100644 index 0000000..2dd5006 --- /dev/null +++ b/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, + IDelayedAccountGrain +{ + private readonly ILogger _logger; + + public DelayedAccountGrain(ILogger logger) + { + _logger = logger; + } + + protected override void OnRegisterRecipients() + { + RegisterRecipient( + nameof(DelayedAccountProjection), + new DelayedAccountProjection(this.GrainFactory, _logger) + ); + } + + public Task GetEvents() => Task.FromResult(Log.Select(e => e.Event).ToArray()); + + public Task GetBalance() => Task.FromResult(ConfirmedState.Balance); + + public Task 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); + } +} diff --git a/src/Strata.Journaling.Tests/JournalingTests/JournaledGrainTests.cs b/src/Strata.Journaling.Tests/JournalingTests/JournaledGrainTests.cs index 1b4075c..9f8f359 100644 --- a/src/Strata.Journaling.Tests/JournalingTests/JournaledGrainTests.cs +++ b/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(accountId); + var projectionGrain = Client.GetGrain(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); + } } \ No newline at end of file diff --git a/src/Strata.Journaling.Tests/JournalingTests/JournalingTestFixture.cs b/src/Strata.Journaling.Tests/JournalingTests/JournalingTestFixture.cs index 26a9ff8..9e045b7 100644 --- a/src/Strata.Journaling.Tests/JournalingTests/JournalingTestFixture.cs +++ b/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(storageProvider); @@ -31,6 +44,7 @@ public JournalingTestFixture() options.CollectionAge = TimeSpan.FromSeconds(61); }); }); + ConfigureTestCluster(builder); Cluster = builder.Build(); } diff --git a/src/Strata.Journaling.Tests/JournalingTests/Projections/AccountProjection.cs b/src/Strata.Journaling.Tests/JournalingTests/Projections/AccountProjection.cs index a3d778f..3fec8d5 100644 --- a/src/Strata.Journaling.Tests/JournalingTests/Projections/AccountProjection.cs +++ b/src/Strata.Journaling.Tests/JournalingTests/Projections/AccountProjection.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; namespace Strata.Journaling.Tests.JournalingTests.Projections; @@ -6,10 +7,12 @@ namespace Strata.Journaling.Tests.JournalingTests.Projections; public sealed class AccountProjection : IOutboxRecipient { 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(accountId); await viewModelGrain.UpdateBalance(balanceEvent.Balance); diff --git a/src/Strata.Journaling.Tests/JournalingTests/Projections/DelayedAccountProjection.cs b/src/Strata.Journaling.Tests/JournalingTests/Projections/DelayedAccountProjection.cs new file mode 100644 index 0000000..c5c2b8c --- /dev/null +++ b/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 +{ + 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(accountId); + await viewModelGrain.UpdateBalance(balanceEvent.Balance); + } + } +} diff --git a/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserCdpGrain.cs b/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserCdpGrain.cs deleted file mode 100644 index e82da33..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserCdpGrain.cs +++ /dev/null @@ -1,8 +0,0 @@ -//using Strata.Sidecars; - -//namespace Strata.Journaling.Tests.SidecarTests.GrainModel; - -//public interface IUserCdpGrain : IGrainWithStringKey, ISidecarGrain -//{ - -//} \ No newline at end of file diff --git a/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserGrain.cs b/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserGrain.cs deleted file mode 100644 index e18780a..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/GrainModel/IUserGrain.cs +++ /dev/null @@ -1,12 +0,0 @@ -//using Strata.Journaling.Tests.SidecarTests.Model; - -//namespace Strata.Journaling.Tests.SidecarTests.GrainModel; - -//public interface IUserGrain : IGrainWithStringKey -//{ -// Task GetData(); - -// ValueTask SetName(string name); - -// ValueTask SetReferenceId(string referenceId); -//} diff --git a/src/Strata.Journaling.Tests/SidecarTests/Grains/UserCdpGrain.cs b/src/Strata.Journaling.Tests/SidecarTests/Grains/UserCdpGrain.cs deleted file mode 100644 index 1ed3d64..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/Grains/UserCdpGrain.cs +++ /dev/null @@ -1,20 +0,0 @@ -//using Strata.Journaling.Tests.SidecarTests.GrainModel; -//using Strata.Sidecars; - -//namespace Strata.Journaling.Tests.SidecarTests.Grains; - -//[GrainType("user-cdp")] -//internal sealed class UserCdpGrain : Grain, IUserCdpGrain -//{ -// public async Task InitializeSidecar() -// { -// // set a reminder, do some work, get the data, etc. - -// var userGrain = GrainFactory.GetGrain( -// this.GetPrimaryKeyString() -// ); - -// await userGrain.SetReferenceId(Guid.NewGuid().ToString()); -// await userGrain.DisableSidecar(); -// } -//} diff --git a/src/Strata.Journaling.Tests/SidecarTests/Grains/UserGrain.cs b/src/Strata.Journaling.Tests/SidecarTests/Grains/UserGrain.cs deleted file mode 100644 index ddf52fc..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/Grains/UserGrain.cs +++ /dev/null @@ -1,36 +0,0 @@ -//using Microsoft.Extensions.DependencyInjection; -//using Strata.Journaling.Tests.SidecarTests.GrainModel; -//using Strata.Journaling.Tests.SidecarTests.Model; -//using Strata.Sidecars; - -//namespace Strata.Journaling.Tests.SidecarTests.Grains; - -//[GrainType("user")] -//internal sealed class UserGrain : Grain, IUserGrain, -// ISidecarHost -//{ -// private readonly IPersistentState _state; - -// public UserGrain( -// [FromKeyedServices("state")] IPersistentState state -// ) -// { -// _state = state; -// } - -// public Task GetData() => -// Task.FromResult(_state.State); - -// public async ValueTask SetName(string name) -// { -// _state.State.Name = name; -// await _state.WriteStateAsync(); -// } - -// public async ValueTask SetReferenceId(string referenceId) -// { -// _state.State.ReferenceId = referenceId; -// await _state.WriteStateAsync(); - -// } -//} diff --git a/src/Strata.Journaling.Tests/SidecarTests/Model/UserData.cs b/src/Strata.Journaling.Tests/SidecarTests/Model/UserData.cs deleted file mode 100644 index 7e8e879..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/Model/UserData.cs +++ /dev/null @@ -1,11 +0,0 @@ -//namespace Strata.Journaling.Tests.SidecarTests.Model; - -//[GenerateSerializer] -//public class UserData -//{ -// [Id(0)] -// public string Name { get; set; } = null!; - -// [Id(1)] -// public string ReferenceId { get; set; } = null!; -//} diff --git a/src/Strata.Journaling.Tests/SidecarTests/SidecarTestFixture.cs b/src/Strata.Journaling.Tests/SidecarTests/SidecarTestFixture.cs deleted file mode 100644 index b9bf96c..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/SidecarTestFixture.cs +++ /dev/null @@ -1,49 +0,0 @@ -//using Microsoft.Extensions.DependencyInjection; -//using Orleans.Journaling; -//using Orleans.TestingHost; -//using Strata.Sidecars; - -//namespace Strata.Journaling.Tests; - -///// -///// Base class for journaling tests with common setup using InProcessTestCluster -///// -//public class SidecarTestFixture : IAsyncLifetime -//{ -// public InProcessTestCluster Cluster { get; } -// public IClusterClient Client => Cluster.Client; - -// public SidecarTestFixture() -// { -// var builder = new InProcessTestClusterBuilder(); -// var storageProvider = new VolatileStateMachineStorageProvider(); - -// builder.ConfigureSilo((options, siloBuilder) => -// { -// siloBuilder.AddSidecars(); -// siloBuilder.AddStateMachineStorage(); -// siloBuilder.Services.AddSingleton(storageProvider); -// }); -// ConfigureTestCluster(builder); -// Cluster = builder.Build(); -// } - -// protected virtual void ConfigureTestCluster(InProcessTestClusterBuilder builder) -// { -// } - -// public virtual async Task InitializeAsync() -// { -// await Cluster.DeployAsync(); -// } - -// public virtual async Task DisposeAsync() -// { -// if (Cluster != null) -// { -// await Cluster.DisposeAsync(); -// } -// } - - -//} diff --git a/src/Strata.Journaling.Tests/SidecarTests/SidecarTests.cs b/src/Strata.Journaling.Tests/SidecarTests/SidecarTests.cs deleted file mode 100644 index 2c724c2..0000000 --- a/src/Strata.Journaling.Tests/SidecarTests/SidecarTests.cs +++ /dev/null @@ -1,22 +0,0 @@ -//using Strata.Journaling.Tests.SidecarTests.GrainModel; - -//namespace Strata.Journaling.Tests.JournalingTests; - -//public class SidecarTests(SidecarTestFixture fixture) : IClassFixture -//{ -// private IGrainFactory Client => fixture.Client; - -// [Fact] -// public async Task Sidecar_ReferenceIdIsSet() -// { -// var grain = Client.GetGrain("user1234"); - -// await Task.Yield(); -// var data = await grain.GetData(); - -// Assert.NotNull(data); -// Assert.NotNull(data.ReferenceId); -// Assert.NotEmpty(data.ReferenceId); -// } - -//} \ No newline at end of file diff --git a/src/Strata.Journaling.Tests/Strata.Journaling.Tests.csproj b/src/Strata.Journaling.Tests/Strata.Journaling.Tests.csproj index 77699f6..54de605 100644 --- a/src/Strata.Journaling.Tests/Strata.Journaling.Tests.csproj +++ b/src/Strata.Journaling.Tests/Strata.Journaling.Tests.csproj @@ -1,18 +1,24 @@  - net9.0 + net10.0 enable enable false - - - - - + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + diff --git a/src/Strata/CHANGELOG.md b/src/Strata/CHANGELOG.md new file mode 100644 index 0000000..7dc5f74 --- /dev/null +++ b/src/Strata/CHANGELOG.md @@ -0,0 +1,5 @@ +## 1.0.1 + +- Updates to .NET 10 +- Updated to Durable State Implementation +- Reworked Outbox diff --git a/src/Strata/EventEnvelope.cs b/src/Strata/EventEnvelope.cs index 2875259..af2b3d8 100644 --- a/src/Strata/EventEnvelope.cs +++ b/src/Strata/EventEnvelope.cs @@ -12,5 +12,5 @@ public class EventEnvelope public int Version { get; set; } [Id(2)] - public DateTime Timestamp { get; set; } + public DateTimeOffset Timestamp { get; set; } } \ No newline at end of file diff --git a/src/Strata/IJournaledGrain.cs b/src/Strata/IJournaledGrain.cs index 3e1df97..ed0dbf4 100644 --- a/src/Strata/IJournaledGrain.cs +++ b/src/Strata/IJournaledGrain.cs @@ -4,6 +4,4 @@ namespace Strata; public interface IJournaledGrain { - [AlwaysInterleave] - Task ProcessOutbox(); } \ No newline at end of file diff --git a/src/Strata/JournaledGrain.cs b/src/Strata/JournaledGrain.cs index 2a98221..97ecc80 100644 --- a/src/Strata/JournaledGrain.cs +++ b/src/Strata/JournaledGrain.cs @@ -1,34 +1,34 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Orleans.Journaling; namespace Strata; public abstract class JournaledGrain : - Grain, IJournaledGrain, ILifecycleParticipant + DurableGrain, IJournaledGrain, ILifecycleParticipant, IRemindable where TModel : new() where TEvent : notnull { - // private IDurableList> _journal = null!; - // private IDurableQueue> _outbox = null!; - // private IPersistentState> _aggregate = null!; - private IPersistentState> _journal = null!; + private const string OutboxCleanupReminderName = "outbox-cleanup"; + + private IDurableList> _journal = null!; + private IDurableQueue> _outbox = null!; + private IPersistentState> _aggregate = null!; private readonly Dictionary> _outboxRecipients = new(); - private IGrainTimer? _outboxTimer = null; + private ILogger _logger = null!; + + private Task? _processOutboxTask; + + private readonly CancellationTokenSource _shutdownCancellationSource = new(); - protected JournaledGrain( - [PersistentState("journal")] IPersistentState> journal - ) - { - _journal = journal; - } #region Lifecycle public void Participate(IGrainLifecycle lifecycle) { lifecycle.Subscribe>( - GrainLifecycleStage.Activate - 1, + GrainLifecycleStage.SetupState - 1, OnHydrateState, OnDestroyState ); @@ -36,40 +36,26 @@ public void Participate(IGrainLifecycle lifecycle) private async Task OnHydrateState(CancellationToken cancellationToken) { - Console.WriteLine("OnHydrateState"); + var loggerFactory = ServiceProvider.GetRequiredService(); + _logger = loggerFactory.CreateLogger(); + + _logger.LogInformation("OnHydrateState"); - //_journal = ServiceProvider.GetRequiredKeyedService<>("journalState"); + _aggregate = ServiceProvider.GetRequiredKeyedService>>("aggregate"); + _journal = ServiceProvider.GetRequiredKeyedService>>("journal"); + _outbox = ServiceProvider.GetRequiredKeyedService>>("outbox"); - if(!_journal.RecordExists) + if (!_aggregate.RecordExists) { - _journal.State = new JournalState(); + _aggregate.State = new() + { + Version = 1 + }; - await _journal.WriteStateAsync(); + await WriteStateAsync(); } - // _aggregate = ServiceProvider.GetRequiredKeyedService>>("aggregate"); - // _journal = ServiceProvider.GetRequiredKeyedService>>("journal"); - // _outbox = ServiceProvider.GetRequiredKeyedService>>("outbox"); - - //if (!_aggregate.RecordExists) - //{ - // _aggregate.State = new(); - // _aggregate.State.Version = 1; - - // await WriteStateAsync(); - //} - OnRegisterRecipients(); - - _outboxTimer = this.RegisterGrainTimer( - ProcessOutbox, - new GrainTimerCreationOptions - { - DueTime = Timeout.InfiniteTimeSpan, - Period = Timeout.InfiniteTimeSpan, - Interleave = true, - } - ); } protected virtual void OnRegisterRecipients() @@ -79,8 +65,39 @@ protected virtual void OnRegisterRecipients() private async Task OnDestroyState(CancellationToken cancellationToken) { + _shutdownCancellationSource.Cancel(); + + if(_processOutboxTask is not null) + { + await _processOutboxTask; + } + /* try to process the outbox */ - await ProcessOutbox(); + if(_outbox is { Count: > 0 }) + { + _logger.LogInformation("Registering reminder for outbox processing."); + await this.RegisterOrUpdateReminder( + OutboxCleanupReminderName, + TimeSpan.FromMinutes(1), + TimeSpan.FromMinutes(1) + ); + } + } + + public async Task ReceiveReminder(string reminderName, TickStatus status) + { + if(reminderName == OutboxCleanupReminderName) + { + _logger.LogInformation("Received reminder for outbox cleanup."); + + await TryProcessOutbox(); + + var reminder = await this.GetReminder(OutboxCleanupReminderName); + if(reminder is not null) + { + await this.UnregisterReminder(reminder); + } + } } #endregion @@ -92,70 +109,39 @@ protected void RegisterRecipient(string key, IOutboxRecipient recipient) #endregion #region Event Processing - protected virtual async Task RaiseEvent(TEvent @event) + protected async Task TryProcessOutbox() { - var newVersion = _journal.State.Version + 1; - - // add it to the log - _journal.State.Events = _journal.State.Events.Union([ - new EventEnvelope { Event = @event, Version = newVersion, Timestamp = DateTime.UtcNow } - ]).ToArray(); - - // apply it to the state - dynamic e = @event!; - dynamic s = _journal.State.Aggregate!; - s.Apply(e); - - // update the version - _journal.State.Version = newVersion; - - // add it to the outbox ... we can loop through a list of providers and add one per provider - // we pass the event and the version, so that the consumer can handle ordering / deduplication - foreach (var recipient in _outboxRecipients.Keys) + if(_processOutboxTask is null || _processOutboxTask.IsCompleted) { - _journal.State.Outbox = _journal.State.Outbox.Union([ - new OutboxEnvelope( - @event, - newVersion, - recipient, - OutboxState.Pending - ) - ]).ToArray(); - //_outbox.Enqueue(new OutboxEnvelope( - // @event, - // newVersion, - // recipient, - // OutboxState.Pending - //)); + _processOutboxTask = ProcessOutbox(); } - - // Save it in one shot - await _journal.WriteStateAsync(); - - // Initialize background processing of the outbox - _outboxTimer?.Change(TimeSpan.FromSeconds(0), Timeout.InfiniteTimeSpan); } - - public async Task ProcessOutbox() + + protected async Task ProcessOutbox() { - _outboxTimer?.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + if(_shutdownCancellationSource.IsCancellationRequested) + { + return; + } - // var failedItems = new List>(); + // It's polite to yield immediately, since we're starting background work. + await Task.CompletedTask.ConfigureAwait(ConfigureAwaitOptions.ForceYielding | ConfigureAwaitOptions.ContinueOnCapturedContext); - var pendingItems = _journal.State.Outbox - .Where(o => o.State == OutboxState.Pending) - .ToList(); + // store the failed items in a list to requeue + var failedItems = new List>(); - foreach(var item in pendingItems) + while(_outbox.TryDequeue(out var item) && !_shutdownCancellationSource.IsCancellationRequested) { + _logger.LogInformation("[{0}] Outbox Count: {1}", this.GetPrimaryKeyString(), _outbox.Count); + try { if (_outboxRecipients.TryGetValue(item.Destination, out var recipient)) { - Console.WriteLine("[{0}] Handling event {1} for recipient {2}", this.GetPrimaryKeyString(), item.Event.GetType().Name, item.Destination); + _logger.LogInformation("[{0}] Handling event {1} for recipient {2}", this.GetPrimaryKeyString(), item.Event.GetType().Name, item.Destination); await recipient.Handle(item.Version, item.Event); - await _journal.WriteStateAsync(); + await WriteStateAsync(); } else { @@ -164,28 +150,65 @@ public async Task ProcessOutbox() } catch (Exception ex) { - Console.WriteLine("[{0}] Failed to handle event {1} for recipient {2}", this.GetPrimaryKeyString(), item.Event.GetType().Name, item.Destination); - // Console.WriteLine(ex.Message + "\r\n" + ex.StackTrace); + _logger.LogError(ex, "[{0}] Failed to handle event {1} for recipient {2}", this.GetPrimaryKeyString(), item.Event.GetType().Name, item.Destination); item.State = OutboxState.Failed; - //failedItems.Add(item); + failedItems.Add(item); } } - //foreach (var item in failedItems) - //{ - // _outbox.Enqueue(item); + foreach (var item in failedItems) + { + _outbox.Enqueue(item); + } - //} + await WriteStateAsync(); + } + + protected virtual async Task RaiseEvent(TEvent @event) + { + var newVersion = _aggregate.State.Version + 1; - await _journal.WriteStateAsync(); + // add it to the log + _journal.Add(new EventEnvelope { + Event = @event, + Version = newVersion, + Timestamp = DateTimeOffset.UtcNow + }); + + // apply it to the state + dynamic e = @event!; + dynamic s = _aggregate.State.Aggregate!; + s.Apply(e); + + // update the version + _aggregate.State.Version = newVersion; + + // add it to the outbox ... we can loop through a list of providers and add one per provider + // we pass the event and the version, so that the consumer can handle ordering / deduplication + foreach (var recipient in _outboxRecipients.Keys) + { + _outbox.Enqueue(new OutboxEnvelope( + @event, + newVersion, + recipient, + OutboxState.Pending + )); + } + + // Save it in one shot + await WriteStateAsync(); + + await TryProcessOutbox(); } #endregion - protected EventEnvelope[] Log => _journal.State.Events; + protected EventEnvelope[] Log => _journal.ToArray(); + + protected bool IsProcessingOutbox => _processOutboxTask is not null && !_processOutboxTask.IsCompleted; - protected TModel ConfirmedState => _journal.State.Aggregate; + protected TModel ConfirmedState => _aggregate.State.Aggregate; - protected int ConfirmedVersion => _journal.State.Version; + protected int ConfirmedVersion => _aggregate.State.Version; } diff --git a/src/Strata/Strata.csproj b/src/Strata/Strata.csproj index 8dd0dbd..83affef 100644 --- a/src/Strata/Strata.csproj +++ b/src/Strata/Strata.csproj @@ -1,7 +1,7 @@  - net9.0 + net10.0 enable enable Strata @@ -15,17 +15,18 @@ orleans;cqrs;event sourcing https://github.com/jsedlak/strata https://github.com/jsedlak/strata - 1.0.0-beta + 1.0.1 - - - + + + - - - + + + +