Cómo actualizar simultáneamente documentos en MongoDb

Question

Estoy trabajando en un sistema distribuido en tiempo real con varios nodos que escriben actualizaciones en la misma colección de MongoDb.

Por simplicidad, el documento se ve así:

{
  id: 1,
  name: foo,
  version: 1
}

En las actualizaciones, la versión está garantizada para aumentar, pero existe la posibilidad de que los mensajes lleguen fuera de orden. Necesito guardar siempre la última versión en la colección.

En SQL haría algo como esto:

  BEGIN TRAN
 
    IF EXISTS ( SELECT * FROM myTable WITH (UPDLOCK) WHERE id= @id )
 
      UPDATE myTable
         SET Name= @Name,
             Version = @Version
       WHERE Id = @id 
       AND Version < @Version
 
    ELSE 
 
      INSERT myTable ( Id, Name, Version )
      VALUES ( @Id, @Name, @Version );
 
  COMMIT

ACTUALIZAR en respuesta a los comentarios: Esto es lo que probé:

await _collection.ReplaceOneAsync(
   x => x.Id == id && x.Version <= payload.Version,
   options: new ReplaceOptions {IsUpsert = true},
   replacement: payload 
);

Al intentar actualizar un documento existente, obtenga esta excepción:

MongoDB.Driver.MongoWriteException: una operación de escritura resultó en un error. E11000 colección de errores de clave duplicada: TEST_948e4e5acae041cca607e9f523dc0194.TestEntityParent índice: id clave dup: {_id: 500}

Answer 1

No estoy seguro de cómo hacer esto con C #. Un ejemplo de actualización $condmediante el shell mongo:

function testUpdate(updated, value) {
    return db.test.update(
             {test: 1},
             [{$set: {
                     monkies: {
                         $cond:{ if: {$or:[
                                      {$gt: [updated, "$version"]},
                                      {$eq: [{$type:"$version"},"missing"]} ]}, then: value, else: "$monkies"
                         }
                     },
                     version: {
                         $cond:{ if: {$or:[
                                      {$gt: [updated, "$version"]},
                                      {$eq: [{$type:"$version"},"missing"]} ]}, then: updated, else: "$version"
                         }
                     }
             }}],
             {upsert: true}
     
    )
}

La primera actualización de una colección vacía no coincide con nada y confirma 1 documento:

replset:PRIMARY> testUpdate(1,1)
WriteResult({
    "nMatched" : 0,
    "nUpserted" : 1,
    "nModified" : 0,
    "_id" : ObjectId("5f6e2116d7a583eef6ad49f9")
})

Una actualización con una versión superior coincide con 1 documento y modifica 1:

replset:PRIMARY> testUpdate(2,3)
WriteResult({ 
     "nMatched" : 1, 
     "nUpserted" : 0, 
     "nModified" : 1 
})

Una actualización con una versión menor coincide con 1 documento, pero actualizaciones 0:

replset:PRIMARY> testUpdate(1,4)
WriteResult({ 
    "nMatched" : 1, 
    "nUpserted" : 0, 
    "nModified" : 0 
})

El documento resultante después de todos estos:

replset:PRIMARY> db.test.find()
{ 
  "_id" : ObjectId("5f6e2116d7a583eef6ad49f9"), 
  "test" : 1, 
  "monkies" : 3, 
  "version" : 2 
}