Optimistic Concurrency and NoSQL

Here is the question: how do i implement optimistic concurrency in this or that document database? How can I make sure that a document being saved has not been changed by someone else? Just to start here are some examples with two popular document stores.

RavenDB

If you need optimistic concurrency you should set the session’s UseOptimisticConcurrency property. Straightforward. Under the hood it uses the ETag values, Raven’s variant of TimeStamp automatically associated with every document. Every time document is updated in the store, its ETag will get a new Guid value. And you get an exception if ETag of the document in the session is not the same as in the store.

[Subject("Raven Experiments")]
public class when_updating_same_object_in_separate_sessions : raven_persistence_context
{
    static Exception Exception;

    Establish context = () => StoreSingleObject(new MyClass {Text = "this one"});
    Because of = () => Exception = Catch.Exception(UpdateTwoObjectsInTwoSessions);
    It should_throw_concurrency_exception = () => Exception.ShouldBeOfType<ConcurrencyException>();

    static void UpdateTwoObjectsInTwoSessions()
    {
        using (var session = Store.OpenSession())
        {
            session.Advanced.UseOptimisticConcurrency = true;
            session
                .Query<MyClass>()
                .First()
                .Text = "still one";

            AtTheSameTimeInAnotherSession();

            session.SaveChanges();
        }
    }

    static void AtTheSameTimeInAnotherSession()
    {
        using (var anotherSession = Store.OpenSession())
        {
            anotherSession
                .Query<MyClass>()
                .First()
                .Text = "ha ha";
            anotherSession.SaveChanges();
        }
    }

    public class MyClass
    {
        public string Id { get; set; }
        public string Text;
    }
}

It works equally good for new documents. For example in event sourcing scenarios sequences of events with continuously incrementing revisions are persisted in the store. The event sequence cannot be consistent if there are two events with the same revision. UseOptimisticConcurrency helps also here. Just make sure that event IDs are generated properly and events with the same Revision have the same ID in the Store.

[Subject("Raven Experiments")]
public class when_storing_two_objects_with_same_id_in_separate_sessions : raven_persistence_context
{
    static Exception Exception;

    Because of = () => Exception = Catch.Exception(CreateTwoObjectsInTwoSessions);
    It should_throw_concurrency_exception = () => Exception.ShouldBeOfType<ConcurrencyException>();

    static void CreateTwoObjectsInTwoSessions()
    {
        using (var session = Store.OpenSession())
        {
            StoreOnce(session, "first");
        }
        using (var session = Store.OpenSession())
        {
            StoreOnce(session, "another first");
        }
    }

    static void StoreOnce(IDocumentSession session, string text)
    {
        session.Advanced.UseOptimisticConcurrency = true;
        session.Store(new MyEvent {Revision = 1, Payload = text});
        session.SaveChanges();
    }

    public class MyEvent
    {
        public int Id { get { return Revision; } }
        public int Revision { get; set; }
        public string Payload { get; set; }
    }
}

Raven design in this case is pretty straightforward and tries to encapsulate details with an out of the box solution. It covers probably 90% of use cases, but if this solution doesn’t really fit to your problem, then you should invent some workarounds.

Taking again the Event Sourcing example. Say you have several sequences of events with ongoing revision numbers. For each sequence there is also some aggregated information like most recent revision and time when the last snapshot was taken. This information will be continuously updated: most recent revision is updated every time new event comes (oft) and snapshot time is set when there is a new snapshot (not so oft). If I update the last revision I do care that nobody is changing it meanwhile (Actually i don’t. It is not a big problem if the last revision not exactly correct. But for the sake of the example let’s assume that I care). If I update the snapshot time, there is no reason to get any conflicts due to changes in revision. That could be a problem with the ETag scenario.

This is where the MongoDB comes in play.

MongoDB

MongoDB has a couple of nice atomic operations: find-and-modify and update-if-current. They are somewhat similar to the Raven’s Patching API with one important difference: in Raven’s partial document updates you must use ID to identify the document that should be patched and ETag to ensure that document has not been changed.

(Sticking to the .NET Driver)

[Subject("Mongo")]
public class when_issuing_an_update_on_a_changed_entity : mongo_context
{
    private static long DocumentsAffected;

    private Establish context = () => SetUpDatabase();
    private Because of = () => DocumentsAffected = UpdateSameObjectTwice();
    private It should_not_affect_any_documents = () => Assert.That(DocumentsAffected, Is.EqualTo(0));

    private static long UpdateSameObjectTwice()
    {
        var events = db.GetCollection("events");
        var oldRevision = events.AsQueryable().First().Revision;

        SomeoneChangedDocument();

        var res = events.Update(
            WhereDocHasRevision(oldRevision),
            Update.Set("Revision", new BsonInt32(100)),
            SafeMode.True);
        return res.DocumentsAffected;

/* or with find-and-modify
        var res = events.FindAndModify(
            WhereDocHasRevision(oldRevision),
            SortBy.Null,
            Update.Set("Revision", new BsonInt32(100)));
        return res.ModifiedDocument == null ? 0 : 1;
*/
    }

    private static void SomeoneChangedDocument()
    {
        var events = db.GetCollection("events");
        var evt = events.AsQueryable().First();
        evt.Revision++;
        events.Save(evt);
    }

    private static IMongoQuery WhereDocHasRevision(int oldRevision)
    {
        return Query.And(
            Query.EQ("_id", "1"),
            Query.EQ("Revision", new BsonInt32(oldRevision))
            );
    }
}