Optimistic Concurrency implementation

[nicomouss]

Dear all,

This post is a question about the current implementation of Optimistic Concurrency in TypeOrm.
I have browsed the source code, but I can't really grasp how Optimistic Concurrency is implemented here. I have provided a simple use-case example below. That is the design pattern I have used countless of time with previous ORMs, like NHibernate for example.

Let's say we have an "Invoice" entity, with an "amount" column. User 1 and User 2 are both editing the same invoice (invoice n°1, having a current amount of 50$, and a version number 12) on their respectives front-end WebApp, and click the save button at the same time:

Chronologically what might happen in our system, considering an optimistic concurrency strategy, are the steps described below:

A - Following the POST request from User 1, our system loads invoice n°1:

SELECT * FROM INVOICE WHERE invoice_id = 1 AND version = 12
=> We get in memory our invoice: { invoice_id:1, amount: 50, version:12 }
=> Business method is called on this invoice object to change its amount to 100$, as requested by User 1

B - Following the POST request from User 2, our system loads the invoice n°1:

SELECT * FROM INVOICE WHERE invoice_id = 1 AND version = 12
=> We get in memory our invoice: { invoice_id:1, amount: 50, version:12 }
=> Business method is called on this invoice object to change its amount to 150$, as requested by User 2

C - At some point, our ORM issues an UPDATE to persist the changes requested by User 1:

UPDATE Invoice 
SET amount = 100, version = version + 1 
WHERE (invoice_id = 1 AND version = 12)
=> Things are Ok here, 1 row has been updated in the database
=> In the database, our Invoice has a new version number of 13

D - At some point, our ORM issues an UPDATE to persist the changes requested by User 2:

UPDATE Invoice 
SET amount = 150, version = version + 1 
WHERE (invoice_id = 1 AND version = 12)
=> 0 row updated in the database (invoice of version number 12 no longer exists in the database, because someone already changed it in the meantime)
=> an OptimisticConcurrencyException is raised by our ORM

E - Our system alerts User 1 that his changes on invoice n°1 has been saved

F - Our system catched the exception raised by our ORM, and alerts User 2 that his changes on invoice n°1 could not be saved, because someone already changed that same invoice in the meantime.

Any idea on how I could implement that design pattern using TypeOrm ?
I have seen the OptimisticLockVersionMismatchError, which is raised while selecting data from the database, but I haven't seen any version checked while Updating data. Maybe I have missed something.

Thanks in advance for any feedback.
Cheers

#Edit: I guess my post here matches post #2848 by @tstearn

[adrianhara]

Yea, not really supported, check this as well #2949

[nicomouss]

Here a solution to fully implement optimistic concurrency with TypeOrm.

This solution uses a subscriber, and is based on entities versioned using an incremented version column (it will not work with entities versioned by a date column).

Process is:

1. Before the update query: we pre-calculate the entity's version we expect to have after the update will be executed.
2. After the update query: we check if the expected version matches the new version of the entity. If the versions does not match, it means the entity had already been updated, and an exception is raised (that you should catch to properly inform the end-user).

The code:

// OptimisticLockingSubscriber.ts
import { EventSubscriber, EntitySubscriberInterface, UpdateEvent } from "typeorm"
import { OptimisticLockVersionMismatchError } from "./OptimisticLockVersionMismatchError"

const EXPECTED_VERSION_METADATA = Symbol();

@EventSubscriber()
export class OptimisticLockingSubscriber implements EntitySubscriberInterface {

    beforeUpdate(event: UpdateEvent<any>) {
        
        // We only deal with entities which have a version number.
        // To know if an entity has a version number, we check if versionColumn 
        // is defined in the metadatas of that entity.
        if (event.metadata.versionColumn) {

            // Getting the current version of the entity
            const currentVersion = Reflect.get(event.entity, event.metadata.versionColumn.propertyName);

            // Calculating the version we expect after the update
            const expectedVersionAfterUpdate = currentVersion + 1;

            // We memorize the expected version as a metadata on the entity
            Reflect.defineMetadata(EXPECTED_VERSION_METADATA, expectedVersionAfterUpdate, event.entity);

        }

    }

    afterUpdate(event: UpdateEvent<any>) {

        // We only deal with entities which have a version number.
        // To know if an entity has a version number, we check if versionColumn 
        // is defined in the metadatas of that entity.
        if (event.metadata.versionColumn) {

            // We retrieve the expected version previously memorized as a metadata on the entity
            const expectedVersion = Reflect.getOwnMetadata(EXPECTED_VERSION_METADATA, event.entity)
            
            // We don't need that metadata anymore, we delete it
            Reflect.deleteMetadata(EXPECTED_VERSION_METADATA, event.entity);

            // Getting the actual version of the entity
            const actualVersion = Reflect.get(event.entity, event.metadata.versionColumn.propertyName);

            // We check if there is version mismatch
            if (expectedVersion != actualVersion) {

                throw new OptimisticLockVersionMismatchError(
                    event.entity,
                    expectedVersion,
                    actualVersion
                )

            }

        }

    }

}

// OptimisticLockVersionMismatchError.ts
export class OptimisticLockVersionMismatchError extends Error {
    name = "OptimisticLockVersionMismatchError";

    constructor(entity: string, expectedVersion: number, actualVersion: number) {
        super();
        Reflect.setPrototypeOf(this, OptimisticLockVersionMismatchError.prototype);
        this.message = `The optimistic lock on entity ${entity} failed, version ${expectedVersion} was expected, but is actually ${actualVersion}.`;
    }

}

You'll need to register the subscriber in the connection options next to your entities:

const options: ConnectionOptions = {
    ...
    entities: [
        ...
    ],
    subscribers: [
        OptimisticLockingSubscriber
    ]
};

Then all your versioned entities like that one:

@Entity()
export class Post {

    @PrimaryGeneratedColumn()
    id: number;

    @Column()
    title: string;

    @Column()
    text: string;

    @VersionColumn() // A versioned entity!
    version: number;
}

will be version-checked automatically on every Save().

[davidc-coder]

@nicomouss awesome solution, love how you've done this as a subscriber. Just a question, you're throwing the error in the afterUpdate, doesn't that mean that in your original scenario User 2 update to amount = 150 will actually save? I can see that an error would be thrown but it seems like the UPDATE will still happen before which isn't ideal?

[nicomouss]

@davidchiew You need to wrap your business logic in a Transaction. Sorry I did not point that out, as it's something that should be systematically done (You can use Transaction decorators on your application's business services to come up with a clean architectural design regarding transactions).

So when an exception is raised somewhere (In this case because of a optimistic concurrency check) then the corresponding transaction is rolled back. In the example, everything done by User 2 would be rolled back, and the UPDATE would not be committed.

[davidc-coder]

@nicomouss thanks that makes sense. Thanks for the tip with the transaction decorator I'll give it a go!

[nicomouss]

@nicomouss thanks that makes sense. Thanks for the tip with the transaction decorator I'll give it a go!

Sure, you can check the related discussion in #1895 if you like

[davidc-coder]

Thanks for the link @nicomouss very interesting discussion, and I'll definitely checkout the new npm package that was created out of it. Apologies I have another question related to your solution above.

How do you get around the fact that event.entity always seems to be undefined inside the beforeUpdate call? It looks like typeorm never actually passes entity in the via the UpdateQueryBuilder.

typeorm/src/query-builder/UpdateQueryBuilder.ts

Line 71 in 10a9e84

queryRunner.broadcaster.broadcastBeforeUpdateEvent(broadcastResult, this.expressionMap.mainAlias!.metadata);

if (this.expressionMap.callListeners === true && this.expressionMap.mainAlias!.hasMetadata) {
    const broadcastResult = new BroadcasterResult();
    queryRunner.broadcaster.broadcastBeforeUpdateEvent(broadcastResult, this.expressionMap.mainAlias!.metadata);
    if (broadcastResult.promises.length > 0) await Promise.all(broadcastResult.promises);
}

[davidc-coder]

Apologies @nicomouss I solved this myself, I wasn't using EntityManager.save(), I was using Repository.update() which seems to call broadcast without providing the entity (line mentioned) earlier. When I call EntityManager.save() update operations appears to broadcast and provide the entity argument.

Thanks for your help! 😄

[nicomouss]

@davidchiew no worries, yes using save() is usually the way to go, as it encapsulates everything.

[davidc-coder]

@nicomouss sorry found an interesting thing, there's an event.databaseEntity property I didn't know about - typeorm appears to load it just before applying the update. Because of this I didn't need the afterUpdate anymore and was able to throw the error on the beforeUpdate which is pretty cool. That way if someone forgets to wrap in a transaction it will still be good!

@EventSubscriber()
export class OptimisticLockingSubscriber implements EntitySubscriberInterface {
  beforeUpdate(event: UpdateEvent<any>) {
    // To know if an entity has a version number, we check if versionColumn
    // is defined in the metadatas of that entity.
    if (event.metadata.versionColumn && event.entity) {
      // Getting the current version of the requested entity update
      const versionFromUpdate = Reflect.get(
        event.entity,
        event.metadata.versionColumn.propertyName
      );

      // Getting the entity's version from the database
      const versionFromDatabase = event.databaseEntity[event.metadata.versionColumn.propertyName];

      // they should match otherwise someone has changed it underneath us
      if (versionFromDatabase !== versionFromUpdate) {
        throw new OptimisticLockVersionMismatchError(
          event.entity,
          versionFromDatabase,
          versionFromUpdate
        );
      }
    }
  }
}

[nicomouss]

@davidchiew No, doing the version check in the beforeUpdate does not makes things better regarding transactional behaviour. If someone has forgotten to wrap the whole business logic within a transaction, then it is just wrong: this has to be fixed if you want to prevent data inconsistencies in your production environment.

Here before your UPDATE, suppose you also had some INSERT done in other tables. Now the UPDATE throws the exception ... then all the previous INSERT also have to be rolled back. The full business logic (INSERT + UPDATE) has to be executed within a unit of work, an atomic operation.

If your architecture is properly designed, there should never be any mistake regarding transactions, as your architectural design should take care of this automatically. Some ideas to do that:

• use Transaction decorator on the entry points of your business actions, as explained in Transaction Management using CLS (Continuation Local Storage) #1895
• use an IoC container which can intercept method calls to your business actions, and thus transparently open/rollback/commit transactions
• Use ES6 proxies to intercept calls to your business actions, and thus transparently open/rollback/commit transactions

[nicomouss]

An example to make this clear.

I have a business action in my system (let's say an API that my mobile client app can use) to add an item to an invoice.

So my API receives the line item to be added to the given invoice. What my system will do is:

1. INSERT a new row into InvoiceItem table
2. UPDATE Invoice row to update the new total of the invoice

For my data to stay consistent, I have to wrap this whole stuff into a Transaction: either the whole INSERT+UPDATE works, either nothing happens (Unit of Work, Atomic operation). My whole system is badly designed if it enables the Item to be added, without the invoice total to be updated. Which is possibly the case if you don't wrap things into a Transaction.

[neuro-sys]

May I ask why don't you use isolation level SERIALIZABLE in the database?

Here's what one of the maintainers say about it.

If you set the transaction isolation level to Serializable, you get what
might be called optimistic concurrency control or optimistic locking
because all concurrent sessions are allowed to proceed in parallel
unless they attempt to alter the same row, in which case one of the
transactions is aborted. So you don't need to check yourself whether a
row was updated; the system will do that for you.

For a discussion from Postgres mailing list about OCC: https://www.postgresql.org/message-id/200501131205.30048.peter_e%40gmx.net

[nicomouss]

May I ask why don't you use isolation level SERIALIZABLE in the database? For a discussion from Postgres mailing list about OCC: https://www.postgresql.org/message-id/200501131205.30048.peter_e%40gmx.net

This implementation is about optimistic concurrency using version number on entities. This approach does not lock anything: you rollback your transaction only if you get a VersionMismatchError. Meaning you are optimistic that there will not be any version mismatch. Usually used with READ_COMMITTED isolation level, it works fine.

You can use SERIALIZABLE, but it's not needed in this case, it will only affect performance.

[nicomouss]

On the other hand, your implementation actually WILL change DB state and only then throw error. Let's say that if we catch error at step 2 (after the update query) we just rollback transaction. Problem is that after step 1 and before rolling back transaction after step 2 our DB actually has wrong data, which could be read and, in very specific case, even be updated over and hence lead to data inconsistency.

Once you rollback a transaction, nothing is persisted in the DB. This is the Unit Of Work design pattern, or atomicity principle: everything works as a whole, or nothing at all.

So here you catch the "OptimisticLockVersionMismatchError" somewhere in you code (up in your layered architecture probably) and you rollback the current active transaction. Thus nothing is persisted in the DB.

[michaelsogos]

An example to make this clear.

I have a business action in my system (let's say an API that my mobile client app can use) to add an item to an invoice.

So my API receives the line item to be added to the given invoice. What my system will do is:

1. INSERT a new row into InvoiceItem table
2. UPDATE Invoice row to update the new total of the invoice

For my data to stay consistent, I have to wrap this whole stuff into a Transaction: either the whole INSERT+UPDATE works, either nothing happens (Unit of Work, Atomic operation). My whole system is badly designed if it enables the Item to be added, without the invoice total to be updated. Which is possibly the case if you don't wrap things into a Transaction.

Dear @nicomouss ,

Regarding properly design the system with Transaction decorator, i completely agree with you.
But about the solution suggested by @davidchiew , i don't get the issue.
I mean, in both case an exception will be raised using pretty same logic, do you see an issue in that code?

[nicomouss]

@michaelsogos I think everything is ok, only thing is @davidchiew was mentioning the following

That way if someone forgets to wrap in a transaction it will still be good!

Which in my opinion should never happen: if you design your architecture with decorators/interceptors for example (= you implement automatic transaction management), there will never be any case of something written in the DB which is not included in a transaction.

Nevertheless if that happens (= something is written in the DB without transaction), it means something is wrongly designed in your architecture.

That was my point.

[gpop-arcadia]

Heres another option if you don't want to force everyone to use save vs update and you dont have to worry about modifying your db configuration as this pushes itself in.

@Injectable()
export class OptimisticLockSubscriber implements EntitySubscriberInterface {

  constructor(
    @InjectConnection() readonly connection: Connection) {
    connection.subscribers.push(this);
  }

 async beforeUpdate(event: UpdateEvent<any>) {
    if (event.metadata.versionColumn && event.entity) {
      const currentVersion = Reflect.get(event.entity, event.metadata.versionColumn?.propertyName);

      let dbVersion = event.databaseEntity?.version;

      if(!dbVersion){
        dbVersion = await this.connection.getRepository(event.metadata.name).findOne(event.entity.id) as any;
        dbVersion = dbVersion[event.metadata.versionColumn.propertyName];
      }

      if (dbVersion !== currentVersion) {
        throw new OptimisticLockVersionMismatchError(event.entity, dbVersion, currentVersion);
      }
    }
  }
}

[imnotjames]

For questions or general discussion, please check out the community slack or check TypeORM's documentation page on other support avenues - cheers!

[falahati]

with the latest version (v0.2.34), @nicomouss solution no longer works as the event.entity is always available but not always has the full entity (in case of an update). Downgrading to (v0.2.33) solves the problem for now and probably the code can be fixed with minor modifications.

[premdasvm]

Is there any way to make this work in the current version? I've been trying to achieve the same, but the event.entity doesn't have the full entity, in my case the version is missing from event.entity but version does get updated on the db, it's just I can't seem to get that value in my subscriber to run my handling logic...

[premdasvm]

Is there any way to make this work in the current version? I've been trying to achieve the same, but the event.entity doesn't have the full entity, in my case the version is missing from event.entity but version does get updated on the db, it's just I can't seem to get that value in my subscriber to run my handling logic...

UPDATE:

On the current version of typeOrm you can use the event.databaseEntity to get the full entity.

so what I did was change the followings from

 const currentVersion = Reflect.get(event.entity, event.metadata.versionColumn.propertyName);

to

const currentVersion = Reflect.get(event.databaseEntity, event.metadata.versionColumn.propertyName);

[falahati]

@premdasvm With v0.3.7, event.databaseEntity is undefined and event.entity is incomplete on beforeUpdate when using manager.update to update a record.

[sparsh-18]

I am using version 0.3.12 with mysql in a nest project and @nicomouss solution with a little tweak works for me.
In the beforeUpdate() function I used event.databaseEntity to get the entire row from the database and in the afterUpdate() function I used the event.entity to get the updated entity by typeorm.

import { BadRequestException } from "@nestjs/common";
import { EventSubscriber, EntitySubscriberInterface, UpdateEvent } from "typeorm"

const EXPECTED_VERSION_METADATA = Symbol();

@EventSubscriber()
export class OptimisticLockingSubscriber implements EntitySubscriberInterface {

    beforeUpdate(event: UpdateEvent<any>) {
    
        if (event.metadata.versionColumn) {

            // Getting the current version of the entity
            const currentVersion = Reflect.get(event.databaseEntity, event.metadata.versionColumn.propertyName);

            // Calculating the version we expect after the update
            const expectedVersionAfterUpdate = currentVersion + 1;

            // We memorize the expected version as a metadata on the entity
            Reflect.defineMetadata(EXPECTED_VERSION_METADATA, expectedVersionAfterUpdate, event.databaseEntity);
        }
    }

    afterUpdate(event: UpdateEvent<any>) {

        if (event.metadata.versionColumn) {
        
            // We retrieve the expected version previously memorized as a metadata on the entity
            const expectedVersion = Reflect.getOwnMetadata(EXPECTED_VERSION_METADATA, event.databaseEntity)
            
            // We don't need that metadata anymore, we delete it
            Reflect.deleteMetadata(EXPECTED_VERSION_METADATA, event.databaseEntity);

            // Getting the actual version of the entity
            const actualVersion = Reflect.get(event.entity, event.metadata.versionColumn.propertyName);

            // We check if there is version mismatch
            if (expectedVersion != actualVersion) {
                // throw error
            }
            
        }
    }

}