OptimisticConcurrencyException is not handled properly in EntityFrameworkReadModelStore

[ryan-nguyen-contemi]

Summary:

I faced the Optimisic Concurrency exception. After skimming through the code, I think that this only happens with EntityFrameworkReadModelStore. I will tell you my understanding and I seek for input from you who are more familiar with the source code.

Description:

I cannot consistently reproduce the issue but it typically happens when multiple commands for the same aggregate ID are published within a short period of time.

Optimistic concurrency exceptions during read model updates might not be handled as expected, potentially leading to data inconsistency (i.e. events saved but not applied to read model).

It appears that reloading the read model from the database upon encountering an optimistic concurrency exception during read model updates might not be happening. This is due to the reuse of the same DbContext instance across retries during read model updates (see this code in EntityFrameworkReadModelStore). Shouldn't there be a DbEntityEntry.Reload() call somewhere to force readmodel reload?

Stacktrace

EventFlow.Exceptions.OptimisticConcurrencyException: The database operation was expected to affect 1 row(s), but actually affected 0 row(s); data may have been modified or deleted since entities were loaded. See http://go.microsoft.com/fwlink/?LinkId=527962 for information on understanding and handling optimistic concurrency exceptions.\n ---> Microsoft.EntityFrameworkCore.DbUpdateConcurrencyException: The database operation was expected to affect 1 row(s), but actually affected 0 row(s); data may have been modified or deleted since entities were loaded. See http://go.microsoft.com/fwlink/?LinkId=527962 for information on understanding and handling optimistic concurrency exceptions.
Npgsql.EntityFrameworkCore.PostgreSQL.Update.Internal.NpgsqlModificationCommandBatch.ConsumeAsync(RelationalDataReader reader, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Update.ReaderModificationCommandBatch.ExecuteAsync(IRelationalConnection connection, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Update.ReaderModificationCommandBatch.ExecuteAsync(IRelationalConnection connection, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Update.Internal.BatchExecutor.ExecuteAsync(IEnumerable1 commandBatches, IRelationalConnection connection, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Update.Internal.BatchExecutor.ExecuteAsync(IEnumerable1 commandBatches, IRelationalConnection connection, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Update.Internal.BatchExecutor.ExecuteAsync(IEnumerable1 commandBatches, IRelationalConnection connection, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.ChangeTracking.Internal.StateManager.SaveChangesAsync(IList1 entriesToSave, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.ChangeTracking.Internal.StateManager.SaveChangesAsync(StateManager stateManager, Boolean acceptAllChangesOnSuccess, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Storage.ExecutionStrategy.<>c__DisplayClass33_02.<<ExecuteAsync>b__0>d.MoveNext()\n--- End of stack trace from previous location ---
Microsoft.EntityFrameworkCore.Storage.ExecutionStrategy.ExecuteImplementationAsync[TState,TResult](Func4 operation, Func4 verifySucceeded, TState state, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Storage.ExecutionStrategy.ExecuteImplementationAsync[TState,TResult](Func4 operation, Func4 verifySucceeded, TState state, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.Storage.ExecutionStrategy.ExecuteAsync[TState,TResult](TState state, Func4 operation, Func4 verifySucceeded, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.DbContext.SaveChangesAsync(Boolean acceptAllChangesOnSuccess, CancellationToken cancellationToken)
Microsoft.EntityFrameworkCore.DbContext.SaveChangesAsync(Boolean acceptAllChangesOnSuccess, CancellationToken cancellationToken)
Komodo.Shared.JsonApi.EventFlowSupport.ReadStores.EntityFrameworkReadModelStore2.UpdateReadModelAsync(TDbContext dbContext, IReadModelContextFactory readModelContextFactory, Func5 updateReadModel, CancellationToken cancellationToken, ReadModelUpdate readModelUpdate) in /src/Shared/src/Komodo.Shared.JsonApi/EventFlowSupport/ReadStores/EntityFrameworkReadModelStore.cs:line 275\n --- End of inner exception stack trace ---
Komodo.Shared.JsonApi.EventFlowSupport.ReadStores.EntityFrameworkReadModelStore2.UpdateReadModelAsync(TDbContext dbContext, IReadModelContextFactory readModelContextFactory, Func5 updateReadModel, CancellationToken cancellationToken, ReadModelUpdate readModelUpdate) in /src/Shared/src/Komodo.Shared.JsonApi/EventFlowSupport/ReadStores/EntityFrameworkReadModelStore.cs:line 275
EventFlow.Core.TransientFaultHandler1.<>c__DisplayClass4_0.<<TryAsync>b__0>d.MoveNext()\n--- End of stack trace from previous location ---
EventFlow.Core.TransientFaultHandler1.TryAsync[T](Func2 action, Label label, CancellationToken cancellationToken)
Komodo.Shared.JsonApi.EventFlowSupport.ReadStores.EntityFrameworkReadModelStore2.UpdateAsync(IReadOnlyCollection1 readModelUpdates, IReadModelContextFactory readModelContextFactory, Func5 updateReadModel, CancellationToken cancellationToken) in /src/Shared/src/Komodo.Shared.JsonApi/EventFlowSupport/ReadStores/EntityFrameworkReadModelStore.cs:line 103
EventFlow.ReadStores.ReadStoreManager2.UpdateReadStoresAsync(IReadOnlyCollection1 domainEvents, CancellationToken cancellationToken)
EventFlow.ReadStores.DispatchToReadStores.<>c__DisplayClass3_0.<<DispatchAsync>b__0>d.MoveNext()\n--- End of stack trace from previous location ---
EventFlow.ReadStores.DispatchToReadStores.<>c__DisplayClass3_0.<<DispatchAsync>b__0>d.MoveNext()\n--- End of stack trace from previous location ---
EventFlow.ReadStores.DispatchToReadStores.DispatchAsync(IReadOnlyCollection1 domainEvents, CancellationToken cancellationToken)
EventFlow.Subscribers.DomainEventPublisher.PublishToReadStoresAsync(IReadOnlyCollection1 domainEvents, CancellationToken cancellationToken)
EventFlow.Subscribers.DomainEventPublisher.PublishAsync(IReadOnlyCollection1 domainEvents, CancellationToken cancellationToken)
EventFlow.Aggregates.AggregateStore.UpdateAsync[TAggregate,TIdentity,TExecutionResult](TIdentity id, ISourceId sourceId, Func3 updateAggregate, CancellationToken cancellationToken)
EventFlow.Aggregates.AggregateStore.UpdateAsync[TAggregate,TIdentity,TExecutionResult](TIdentity id, ISourceId sourceId, Func3 updateAggregate, CancellationToken cancellationToken)
MyDecoratedCustomLoggingAggregateStore.UpdateAsync[TAggregate,TIdentity,TExecutionResult](TIdentity id, ISourceId sourceId, Func3 updateAggregate, CancellationToken cancellationToken) in MyDecoratedCustomLoggingAggregateStore.cs
EventFlow.CommandBus.ExecuteCommandAsync[TAggregate,TIdentity,TResult](ICommand3 command, CancellationToken cancellationToken)
EventFlow.CommandBus.PublishAsync[TAggregate,TIdentity,TResult](ICommand3 command, CancellationToken cancellationToken)

[rasmus]

I'm personally not an EF expert. But wouldn't these line update the value?

EventFlow/Source/EventFlow.EntityFramework/ReadStores/EntityFrameworkReadModelStore.cs

Lines 253 to 255 in b277c52

	var databaseValues = await entry.GetDatabaseValuesAsync(cancellationToken)
	.ConfigureAwait(false);
	entry.CurrentValues.SetValues(databaseValues);

[rasmus]

You could also try providing a custom implementation for this class https://github.com/eventflow/EventFlow/blob/b277c52ea409c52e5f5b78f9cedfd0b3d399015d/Source/EventFlow/Core/RetryStrategies/OptimisticConcurrencyRetryStrategy.cs that adds some metrics to get some insights into the amount of times its happening. For a high load system it might simply be because the retry limit is reached.