Add Local Provider optimistic concurrency support #1414
Conversation
| describe('With two projections for the same ReadModel', () => { | ||
| if (process.env.TESTED_PROVIDER === 'AWS') { | ||
| console.log('AWS Provider is not working properly when inserting a ReadModel with two projections') // TODO: Fix AWS Provider | ||
| return |
There was a problem hiding this comment.
We should create an issue to track this and add details
| docs.map((doc: ReadModelEnvelope) => { | ||
| doc.value.boosterMetadata!.optimisticConcurrencyValue = this.generateOptimisticConcurrencyValue(doc.value) | ||
| }) | ||
| resolve(docs) |
There was a problem hiding this comment.
Why do we need to do this when reading the read model? Would it not be already stored in the DB?
There was a problem hiding this comment.
There could be old readmodels where the value is undefined I removed the field, see my next comment
| const readModelOptimistic: ReadModelEnvelope & { uniqueKey?: string } = readModel | ||
| readModelOptimistic.uniqueKey = `${readModel.typeName}_${readModel.value.id}_${readModel.value.boosterMetadata?.version}` | ||
| if (readModelOptimistic.value.boosterMetadata?.version === 1) { | ||
| return this.insert(readModel) | ||
| } | ||
| return this.update(readModelOptimistic, expectedCurrentVersion) |
There was a problem hiding this comment.
So we have three values (I think) related to optimistic concurrency, but I don't fully get their purpose. We have:
- readModel.boosterMetadata.version (contains the read model version)
- readModel.uniqueKey (a unique key build from the version so that you can achieve the optimistic concurrency)
- readModel.boosterMetadata.optimisticConcurrencyValue (apparently, it also contains the read model version?)
Could you please clarify their purpose?
There was a problem hiding this comment.
You are right, the optimisticConcurrencyValue is not needed as it will be the version always. I will remove it and simplify it. The uniqueKey is, as you said, used to avoid inserting same ReadModel twice. The value of this field will be <id>_<boosterMetadata.version>. If the ReadModel doesn't exists in the database and there are two instances trying to insert the same ReadModel, then both of them will try to do it with version = 0. To be sure that only one of the insert finish successfully we will need to look up into the database before inserting to check if another instance inserted the same read model. As we don't are not able to lock rows or tables we need a unique key to ensure we will never insert a read model with the same and same version (in this case 0).
| 'value.id': readModel.value.id, | ||
| $or: [ | ||
| { 'value.boosterMetadata.optimisticConcurrencyValue': { $exists: false } }, | ||
| { 'value.boosterMetadata.optimisticConcurrencyValue': expectedCurrentVersion }, |
There was a problem hiding this comment.
Oh! I thought NeDB didn't support optimistic concurrency... Why do we need this condition if we are already using the unique key approach?
There was a problem hiding this comment.
I'll change it to use the version as the field is not needed. About your question, the uniqueKey is only to prevent inserting the same read model twice and the version to update only when the version number is equals to the expected one. If the version is not equals to the expected one it means that the row was updated by another instance.
| (err) => { | ||
| { upsert: false, returnUpdatedDocs: true }, | ||
| (err: unknown, numAffected: number) => { | ||
| if (numAffected === 0) { |
There was a problem hiding this comment.
I though when we have an error updating something, numAffected is always 0. How do you know that if this condition is true, then it must be a optimistic concurrency error?
There was a problem hiding this comment.
Yes, it's not a perfect solution but in this case we are updating read models only son we can expect that the updated documents will be greater than 0. In this case, using the query (version = expected version) while updating we will have 0 documents updated if the row was updated by another instance.
Add rush change Add local provider concurrency
gonzalojaubert
force-pushed
the
local_concurrency
branch
from
bd7b7dd
to
a0aca17
Compare
|
/integration sha=a0aca17 |
|
⌛ Integration tests are running... Check their status here 👈 |
|
✅ Integration tests have finished successfully! |
Add Local Provider optimistic concurrency support