How To Handle Concurrency in NEventStore

[wadsworj]

I'm wondering if anyone can give some pointers when it comes to handling concurrency in NEventStore.

We are using NEventStore where multiple clients can insert events into the Commits table at the same time.

Initially I was using EnlistInAmbientTransaction() with TransactionScope but this resulted in multiple deadlocks. Turning off EnlistInAmbientTransaction() and removing the TransactionScope greatly reduced the number of deadlocks, however now I am getting NEventStore.ConcurrencyException.

How am I supposed to be handling this exception?

Some code

return Wireup.Init()
                         .LogToOutputWindow()
                         .UsingSqlPersistence(connectionFactory)
                         .WithDialect(new MsSqlDialect())
                         //.EnlistInAmbientTransaction() // two-phase commit
                         .InitializeStorageEngine()
                         .UsingJsonSerialization()
                         .Compress()
                         .EncryptWith(EncryptionKey)
                         .HookIntoPipelineUsing(new[] { new AuthorizationPipelineHook() })
                         .Build();

---

using (store = WireupEventStore())
            {
                AppendToStream(@event, streamId);
            }

private void AppendToStream(IEvent @event, Guid streamId)
        {
            using (IEventStream stream = store.OpenStream(@event.EventType.ToString(), streamId))
            {
                stream.Add(new EventMessage { Body = @event });
                try
                {
                    stream.CommitChanges(Guid.NewGuid());
                }
                catch (NEventStore.ConcurrencyException)
                {
                    Thread.Sleep(100);

                    try
                    {
                        // can't insert event
                        stream.CommitChanges(Guid.NewGuid());
                    }
                    catch (NEventStore.ConcurrencyException ex)
                    {
                        Defect.Log(ex);
                    }
                }
            }
        }

[fschmied]

A ConcurrencyException at this point means that someone else modified exactly this stream between your calls to "OpenStream" and "CommitChanges".If you don't care, you can just reopen the stream and try again. Performance-wise it would be nice if one could tell NEventStore to just append a message and allow concurrent writes to the same stream (ideally, I'd like to be able to append messages without even reading the rest of the stream), but I don't think there is an API for this. Best regards, Fabian

…

On Wed, Mar 13, 2019 at 3:39 PM James Wadsworth ***@***.***> wrote: I'm wondering if anyone can give some pointers when it comes to handling concurrency in NEventStore. We are using NEventStore where multiple clients can insert events into the Commits table at the same time. Initially I was using EnlistInAmbientTransaction() with TransactionScope but this resulted in multiple deadlocks. Turning off EnlistInAmbientTransaction() and removing the TransactionScope greatly reduced the number of deadlocks, however now I am getting NEventStore.ConcurrencyException. How am I supposed to be handling this exception? Some code return Wireup.Init() .LogToOutputWindow() .UsingSqlPersistence(connectionFactory) .WithDialect(new MsSqlDialect()) //.EnlistInAmbientTransaction() // two-phase commit .InitializeStorageEngine() .UsingJsonSerialization() .Compress() .EncryptWith(EncryptionKey) .HookIntoPipelineUsing(new[] { new AuthorizationPipelineHook() }) .Build(); ------------------------------ using (store = WireupEventStore()) { ***@***.*** <https://github.com/event>, streamId); } private void AppendToStream(IEvent @event <https://github.com/event>, Guid streamId) { using (IEventStream stream = ***@***.***(), streamId)) { stream.Add(new EventMessage { Body = @event <https://github.com/event> }); try { stream.CommitChanges(Guid.NewGuid()); } catch (NEventStore.ConcurrencyException) { Thread.Sleep(100); try { // can't insert event stream.CommitChanges(Guid.NewGuid()); } catch (NEventStore.ConcurrencyException ex) { Defect.Log(ex); } } } } — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#472>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAIhLADa3tJCxQn_gtJJiCW_gXjGC_Mvks5vWQ2hgaJpZM4btZeD> .

[AGiorgetti]

As @fschmied explained, you get concurrency exception because multiple streams (opened on the same stramid) modify the data at the same time.

At this point there's no Api that allows you to write at the end of the stream without opening it (I admin that it can be useful in some case).

But you look at how the CommitChanges is implemented in the OptimisticEventStream you'll see that whenever a ConcurrencyException is thrown the new events are read from the store and the stream is updated.

At this point you have all your uncommited events at the end of the updated stream and you can try to commit them again (excatly as you did in your code fragment). Whether this behavior is right or wrong it depends on your domain logic.

If the new events can be safely appended at the end of the stream just save the stream again, if not notify the used are open a new stream on the same id.

Anyhow it mostly depends on your scenario and what you are trying to achieve, why do you need to modify the same stream from different threads?

[AGiorgetti]

I'm aware that there's a problem when trying to commit events in the middle of a stream, we throw a ConcurrencyException there and we exit without refreshing the stream, breaking the "normal" behavior of the eventstream, will change that before the official v6 release.